Skip to content

Getting Started

Installation

AegisMCP is published on PyPI. You can install the minimal core, which has zero heavy dependencies and runs entirely on anyio and pydantic.

pip install aegismcp-core

Or, install with all optional extensions (HTTP transports, observability SDKs, vector database adapters):

pip install "aegismcp-core[all]"

Your First MCP Server (Stdio)

The simplest AegisMCP application runs entirely over standard input/output (Stdio). This is the standard way MCP clients communicate with local servers.

Create a file called server.py:

import asyncio
from aegismcp.server.app import AegisMCP

# 1. Initialize the application
app = AegisMCP("HelloWorld")

# 2. Register tools
@app.tool(description="Say hello to a user")
async def say_hello(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    # 3. Start the Stdio transport
    asyncio.run(app.run_stdio())

You can now connect any standard MCP Client to this server by pointing it to execute python server.py.

Advanced Tool Definition

Unlike toy frameworks, AegisMCP deeply integrates with Pydantic to provide robust, complex input validation out of the box.

Pydantic Models as Inputs

You can pass complex nested structures directly into your tool signature, and AegisMCP will automatically extract the JSON schema to send to the LLM.

from pydantic import BaseModel, Field

class UserQuery(BaseModel):
    query: str = Field(..., description="The exact SQL query to execute")
    dry_run: bool = Field(False, description="If true, simulates the execution")
    timeout_ms: int = Field(5000, ge=1000, le=30000)

@app.tool(description="Executes a database query")
async def execute_query(payload: UserQuery) -> str:
    if payload.dry_run:
        return f"Would have executed: {payload.query}"

    # Execute actual query...
    return "Execution successful"

Sync vs Async Execution

AegisMCP prevents you from accidentally blocking your server.

If you write a def function (synchronous), AegisMCP will automatically offload it to a thread pool via asyncio.get_running_loop().run_in_executor().

import time

@app.tool(description="A heavy CPU task")
def heavy_cpu_task(iterations: int) -> str:
    # This will NOT block the main async event loop
    time.sleep(5) 
    return "Done"

@app.tool(description="A fast network task")
async def fast_network_task() -> str:
    # This runs directly on the async event loop
    await asyncio.sleep(0.1)
    return "Done"