aiomcp 0.0.12

A Simple Python MCP Solution

aiomcp

A Simple Python MCP Solution

Mission of aiomcp

Start with a smooth experience, end with a compliant solution.

Quick Start

A Simple STDIO Example

MCP server server.py

import asyncio

from aiomcp import McpServer, McpStdioServerTransport


async def add(a: int, b: int) -> int:
    return a + b


async def main() -> None:
    mcp_server = McpServer("mcp-server-name")
    await mcp_server.register_tool(add)
    await mcp_server.host(McpStdioServerTransport())


if __name__ == "__main__":
    asyncio.run(main())

MCP client client.py

import asyncio
import sys

from aiomcp import McpClient, McpStdioClientTransport


async def main() -> None:
    mcp_client = McpClient("mcp-client-name")
    await mcp_client.initialize(
        McpStdioClientTransport([sys.executable, "server.py"])
    )

    try:
        sum_value = await mcp_client.invoke("add", {"a": 1, "b": 2})
        print(sum_value)
        assert sum_value == 3
    finally:
        await mcp_client.close()


if __name__ == "__main__":
    asyncio.run(main())

Run the client from the same directory as server.py:

python client.py

A Simple HTTP Example

MCP server server.py

import asyncio

from aiomcp import McpServer


async def add(a: int, b: int) -> int:
    return a + b


async def main() -> None:
    mcp_server = McpServer("mcp-server-name")
    await mcp_server.register_tool(add)
    await mcp_server.host("http://127.0.0.1:8000/mcp")


if __name__ == "__main__":
    asyncio.run(main())

Run the server, it will host on http://127.0.0.1:8000/mcp

python server.py

MCP client client.py

import asyncio

from aiomcp import McpClient


async def main() -> None:
    mcp_client = McpClient("mcp-client-name")
    await mcp_client.initialize("http://127.0.0.1:8000/mcp")

    try:
        sum_value = await mcp_client.invoke("add", {"a": 1, "b": 2})
        assert sum_value == 3
    finally:
        await mcp_client.close()


if __name__ == "__main__":
    asyncio.run(main())

Run the client from anywhere in the same machine:

python client.py

A Complete LLM-Friendly HTTP MCP Server

For best compatibility with LLM clients, tools can return an array of MCP content blocks. Each item in the array is one content block, such as text, image, audio, resource link, or embedded resource.

import asyncio

from aiomcp import McpServer


async def screenshot():
    image_data = get_screenshot_image_data()

    return [
        {"type": "text", "text": "Here is the captured screenshot."},
        {"type": "image", "data": image_data, "mimeType": "image/png"},
    ]


async def main() -> None:
    mcp_server = McpServer("computer-use-server")
    await mcp_server.register_tool(screenshot)
    await mcp_server.host("http://127.0.0.1:8000/mcp")


if __name__ == "__main__":
    asyncio.run(main())

Server Tool Registration

Register Python Callables as Tools

register_tool accepts ordinary Python callables. Use a function, async function, bound method, class method, or static method; aiomcp reads the type hints and exposes the callable as an MCP tool.

def add(a: int, b: int) -> int:
    return a + b


async def fetch_status(service: str) -> str:
    return f"{service}: ok"


class Calculator:
    def scale(self, value: float, factor: float = 1.0) -> float:
        return value * factor

    @classmethod
    async def add_offset(cls, value: int, offset: int) -> int:
        return value + offset

    @staticmethod
    async def total(items: list[int]) -> int:
        return sum(items)


calculator = Calculator()

await mcp_server.register_tool(add)
await mcp_server.register_tool(fetch_status)
await mcp_server.register_tool(calculator.scale, alias="calculator_scale")
await mcp_server.register_tool(Calculator.add_offset)
await mcp_server.register_tool(Calculator.total)

Clients invoke registered tools by name with a dict of arguments, similar to Python keyword arguments.

assert await mcp_client.invoke("add", {"a": 1, "b": 2}) == 3
assert await mcp_client.invoke("calculator_scale", {"value": 3, "factor": 2}) == 6.0
assert await mcp_client.invoke("total", [1, 2, 3]) == 6

Customize Parameters and Schemas

Use Annotated with Pydantic Field for parameter descriptions. Use Pydantic models when a tool takes a structured object.

from enum import StrEnum
from typing import Annotated

from pydantic import BaseModel, Field


class SearchRequest(BaseModel):
    query: str
    limit: int = 5


class Tone(StrEnum):
    WARM = "warm"
    PRECISE = "precise"


async def search(
    request: SearchRequest,
    tone: Annotated[Tone, Field(description="Response tone")] = Tone.PRECISE,
) -> list[str]:
    return [f"{tone.value}: {request.query}"][: request.limit]


await mcp_server.register_tool(
    search,
    description="Search documents for the current workspace.",
    annotations={"readOnlyHint": True, "openWorldHint": False},
)

Advanced Tool Registration

Use mcp_tools_register when you already have JSON schemas or want exact control over the tool contract.

def add(a: int, b: int) -> int:
    return a + b


await mcp_server.mcp_tools_register(
    "add",
    add,
    input_schema={
        "type": "object",
        "properties": {
            "a": {"type": "integer", "description": "The first number"},
            "b": {"type": "integer", "description": "The second number"},
        },
        "required": ["a", "b"],
    },
    output_schema={"type": "integer"},
    title="Add Numbers",
    icons=[{"src": "https://example.com/add.png", "mimeType": "image/png"}],
)

MCP Transports

Using McpTransports

While McpClient and McpServer handle client/server behavior, they are implemented in a platform-agnostic way by using McpClientTransport/McpServerTransport to handle message streams. Client and server transports usually come in pairs as McpTransport implementations.

mcp_server = McpServer()
await mcp_server.host(McpHttpTransport("127.0.0.1", 8000, "/mcp"))
# await mcp_server.host("http://127.0.0.1:8000/mcp")

mcp_client = McpClient()
await mcp_client.initialize(McpHttpTransport("127.0.0.1", 8000, "/mcp"))
# await mcp_client.initialize("http://127.0.0.1:8000/mcp")

Supported McpTransport implementations:

• McpDirectTransport: Simple client driven transport that invokes McpServer with minimal MCP protocol overhead.
• McpMemoryTransport: Single event loop, in-memory transport.
• McpHttpTransport
• McpStdioClientTransport / McpStdioServerTransport: stdio based transport

Sample: Using McpMemoryTransport for local debugging.

transport = McpMemoryTransport()

mcp_server = McpServer()
asyncio.create_task(mcp_server.host(transport))

mcp_client = McpClient()
await mcp_client.initialize(transport)

Tool Results and LLM Integration

MCP Tool Result and Invoke Options

The MCP specification defines tool results with both content and structuredContent. The content field should be a list of text, image, audio, resource link, or embedded resource blocks. The optional structuredContent field is a structured object that follows the tool's output schema when one exists.

McpClient.invoke(...) accepts per-call result shaping options.

• use_structured_content: return structuredContent only, ignoring content.
• convert_mcp_tool_result_content_format: return MCP content blocks, parsing both content and structuredContent as MCP content or lists of MCP content when possible.

LLM Integration Client Best Practice

As an MCP client, aiomcp follows the MCP specification and uses content and structuredContent as is by default.

If an MCP server does not produce standard MCP content blocks, aiomcp can optionally convert and merge content and structuredContent into an LLM-friendly content list with convert_mcp_tool_result_content_format, making it optimal for LLM integration.

from aiomcp import McpClient, McpInvokeError


# This function returns a payload suitable for most LLM API tool results.
async def call_mcp_tool(
    client: McpClient,
    tool_call_id: str,
    tool_name: str,
    arguments: dict,
) -> dict:
    try:
        output = await client.invoke(
            tool_name,  # such as "screenshot"
            arguments,  # such as {"window": "active"}
            convert_mcp_tool_result_content_format=True,
        )
    except McpInvokeError as error:
        # The complete raw MCP result is available for logging or recovery.
        raw_result = error.result
        output = raw_result.model_dump_json(exclude_none=True, by_alias=True)

    # Use the format that fits best to your LLM API.
    return {
        "role": "tool",
        "tool_call_id": tool_call_id,
        "content": output,
    }

For custom tool result handling, use McpClient.invoke_result(...) to get the full McpCallToolResult.

LLM Integration Server Best Practice

As an MCP server, aiomcp generates output schemas from return type hints by default. Some MCP clients work best with content-first results. For broad LLM compatibility, omit the return annotation or set skip_mcp_tool_output_schema, then return a list of MCP content blocks. A simple text block can be a dict like {"type": "text", "text": "..."} or a McpTextContent; richer responses can use McpImageContent, McpAudioContent, McpResourceLink, or McpEmbeddedResource.

from aiomcp import McpCallToolResult, McpImageContent, McpServer, McpTextContent


mcp_server = McpServer(flags={"skip_mcp_tool_output_schema": True})


async def use_dict_content():
    return [
        # This follows the MCP content contract, not any provider-specific LLM API shape.
        {"type": "text", "text": "Chart generated."},
        {"type": "image", "data": "base64-image-data", "mimeType": "image/png"},
    ]


async def use_mcp_contracts():
    return [
        McpTextContent(text="Chart generated."),
        McpImageContent(data="base64-image-data", mimeType="image/png"),
    ]


async def use_call_tool_result():
    return McpCallToolResult(
        content=[McpTextContent(text="Chart generated.")],
        structuredContent={"chartId": "revenue-q1"},
    )

Authorization

Connect to an OAuth 2.1 protected remote MCP server.

from aiomcp import McpClient, McpAuthorizationClient

authorization = await McpAuthorizationClient.discover("http://remote-server/mcp")

# Or use a bearer token
# authorization = McpAuthorizationClient("your-access-token")

client = McpClient("mcp-client-name")
await client.initialize("http://remote-server/mcp", authorization=authorization)

Use GitHub as the OAuth provider to connect to GitHub's MCP server.

from aiomcp import McpClient, McpAuthorizationClient

server_url = "https://api.githubcopilot.com/mcp/"

authorization = await McpAuthorizationClient.discover(
    server_url,
    client_id="{YOUR_CLIENT_ID}",
    client_secret="{YOUR_CLIENT_SECRET}",
)

client = McpClient("mcp-client-name")
await client.initialize(server_url, authorization=authorization)

Protect your MCP server with built-in OAuth 2.1 + Dynamic Client Registration.

from aiomcp import McpServer, McpAuthorizationServer

server = McpServer("mcp-server-name")
await server.register_tool(add, alias="add")

authorization = McpAuthorizationServer()
await server.host("http://127.0.0.1:8000/mcp", authorization=authorization)

Compatibility flags

aiomcp defaults to broad compatibility, so it can connect to a wider range of MCP clients and servers. Compatibility flags let you opt into stricter protocol checks when you want closer MCP enforcement.

from aiomcp import McpClient, McpServer

mcp_client = McpClient(
    flags={"enforce_mcp_tool_result_content": True}
)

mcp_server = McpServer(
    flags={"enforce_mcp_initialize_sequence": True}
)

Available client flags

• throw_mcp_contract_errors: raise error when incoming MCP messages miss required contract fields.
• throw_mcp_parse_errors: raise when a transport receives malformed MCP JSON-RPC.
• enforce_mcp_tools_capability: require the server to advertise the tools capability before the client sends tools/list.
• enforce_mcp_tool_result_content: reject tool results that omit content.
• enforce_mcp_version_negotiation: reject unsupported negotiated protocol versions.
• enforce_mcp_session_header: require HTTP session headers where applicable.
• enforce_mcp_protocol_header: require HTTP protocol version headers where applicable.
• enforce_mcp_transport_version_consistency: require HTTP header and initialize body protocol versions to match.

Available server flags

• throw_mcp_parse_errors: raise when a transport receives malformed MCP JSON-RPC instead of ignoring it.
• enforce_mcp_initialize_sequence: reject requests sent before MCP initialization completes.
• enforce_mcp_version_negotiation: negotiate only supported protocol versions.
• enforce_mcp_session_header: require HTTP session headers where applicable.
• enforce_mcp_protocol_header: require HTTP protocol version headers where applicable.
• skip_mcp_tool_output_schema: skip output schema generation for registered tools.
• enforce_mcp_tool_result_content_format: validate tool result content against MCP content block.
• allow_mcp_tool_result_empty_content: allow tool results to omit the content.