T17 — MCP stdio + HTTP transport với dynamic adapter

{`
class McpClientManager:
    def __init__(self, configs: list[McpServerConfig]):
        self._configs = configs
        self._sessions: dict[str, ClientSession] = {}
        self._exit_stack = AsyncExitStack()

    async def start(self) -> None:
        for config in self._configs:
            session = await self._connect(config)
            self._sessions[config.name] = session

    async def _connect(self, config: McpServerConfig) -> ClientSession:
        if config.transport == "stdio":
            transport = await self._exit_stack.enter_async_context(
                stdio_client(StdioServerParameters(
                    command=config.command,
                    args=config.args,
                    env=config.env,
                ))
            )
        else:  # streamable_http
            transport = await self._exit_stack.enter_async_context(
                streamablehttp_client(config.url, headers=config.headers)
            )
        session = await self._exit_stack.enter_async_context(
            ClientSession(*transport)
        )
        await session.initialize()
        return session

    async def list_all_tools(self) -> list[Tool]:
        """Aggregate tools from all servers into flat namespace."""
        tools = []
        for name, session in self._sessions.items():
            result = await session.list_tools()
            for tool in result.tools:
                tools.append(ToolWithServer(
                    tool=tool,
                    server_name=name,
                ))
        return tools

    async def stop(self) -> None:
        await self._exit_stack.aclose()
`}

{`
def build_dynamic_pydantic_model(tool_schema: dict) -> type[BaseModel]:
    """Create runtime Pydantic model from MCP tool JSON schema."""
    fields = {}
    properties = tool_schema.get("properties", {})
    required = set(tool_schema.get("required", []))

    for field_name, field_schema in properties.items():
        python_type = _json_type_to_python(field_schema)
        if field_name in required:
            fields[field_name] = (python_type, ...)
        else:
            default = field_schema.get("default", None)
            fields[field_name] = (python_type | None, default)

    return create_model("DynamicToolInput", **fields)
`}

{`
# Kịch bản: my_tool_server process crash (OOM, exception, etc.)
# AsyncExitStack cleanup fires cho server đó
# Các session khác trong stack KHÔNG bị ảnh hưởng

# "AsyncExitStack ensures each MCP server's lifecycle is managed
#  independently — one server crash doesn't cascade"

# Vấn đề: _sessions["my_tool"] vẫn tồn tại trong dict
# → Agent cố gọi tool → session đã dead → RuntimeError
# Fix cần: health check hoặc reconnect logic trong _connect()
`}

{`
# MCP server trả về schema với type "anyOf" hoặc nested $ref
tool_schema = {
    "properties": {
        "config": {
            "anyOf": [{"type": "string"}, {"type": "object"}]  # union type
        }
    }
}

# _json_type_to_python() không handle "anyOf"
# → KeyError hoặc fallback về Any
# → Tool vẫn xuất hiện nhưng với type Any (mất validation)
# → Hoặc Exception → tool invisible (không được add vào list_all_tools)

# Fix: handle "anyOf" → Union type, "$ref" → resolve schema
`}

{`
import asyncio
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def minimal_mcp_manager(configs):
    """Minimal McpClientManager — connects, aggregates, cleans up."""
    stack = AsyncExitStack()
    sessions = {}

    async with stack:
        for cfg in configs:
            params = StdioServerParameters(
                command=cfg["command"],
                args=cfg.get("args", []),
            )
            transport = await stack.enter_async_context(stdio_client(params))
            session = await stack.enter_async_context(ClientSession(*transport))
            await session.initialize()
            sessions[cfg["name"]] = session

        # Aggregate tools from all servers
        all_tools = []
        for name, session in sessions.items():
            result = await session.list_tools()
            for tool in result.tools:
                all_tools.append({"server": name, "tool": tool})

        return all_tools
        # AsyncExitStack cleans up all sessions on exit

# Usage:
# configs = [
#     {"name": "git", "command": "python", "args": ["-m", "git_mcp"]},
#     {"name": "search", "command": "npx", "args": ["@search/mcp-server"]},
# ]
# tools = asyncio.run(minimal_mcp_manager(configs))
`}