aiomcp 0.0.5

A Simple Python MCP Solution

aiomcp

A Simple Python MCP Solution

Mission of aiomcp

A smooth start experience.

Tutorial

Expected Behavior for Version 0.1.0

A simple McpServer

def func(a: int, b: int):
    return a + b

impl = ToolImpl()

mcp_server = McpServer("mcp-server-name")
await mcp_server.register_tool(func)
await mcp_server.register_tool(impl.method, alias="tool_alias")
await mcp_server.register_tool(ToolImpl.class_method)
await mcp_server.host("http://127.0.0.1:8000/mcp")

A simple McpClient

mcp_client = McpClient("mcp-client-name")
await mcp_client.initialize("http://127.0.0.1:8000/mcp")
# await mcp_client.initialize(McpHttpTransport("127.0.0.1", 8000, "/mcp"))
# await mcp_client.initialize(McpStdioClientTransport([sys.executable, "server.py"]))
# await mcp_client.initialize(mcp_server) for quick testing without hosting a server
sum_value = await mcp_client.invoke("name", {"a": 1, "b": 2})
# standard MCP functions
# await mcp_client.mcp_tools_list() -> list[McpTool]
# await mcp_client.mcp_tools_call(tool: McpTool, request: McpRequest) -> McpResponse | McpError

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)