Writing a new adapter or transport

Section 9 · From docs/MCP_INTERNALS.md on GitHub

There are two ways in, depending on whether the target speaks MCP itself.

A framework that dispatches MCP methods to your code (another SDK, a gateway): follow lowlevel.py. For each method it hands you, decide ownership, build a ToolContext with layer.context(principal, request_id=..., progress_callback=..., log_callback=..., session=...), call the layer method, convert to_wire() output to the framework's types, and map MCPLayerError to the framework's error type (NotFoundError to -32602, or -32002 for resources on handshake-era connections). Wrap the host's existing handlers instead of replacing them, and merge list results. Forward change events with a ChangeFanout. export_definitions already gives you per-method handlers if the framework is simple enough.

A raw transport (a socket, a message queue, a websocket): reuse the native protocol engine and write only the framing. One Session per connection; feed it decoded messages; send back what handle returns; pass a send coroutine for server-initiated notifications. This TCP transport is complete and was run:

"""A newline-delimited JSON-RPC transport over TCP for NativeMCPServer."""
import asyncio
import json

from cloudg.mcp import CloudGMCPLayer, Principal
from cloudg.mcp.native import NativeMCPServer
from cloudg.mcp.native.jsonrpc import PARSE_ERROR, dumps, error_response


async def serve_tcp(server: NativeMCPServer, host: str = "127.0.0.1", port: int = 0):
    async def on_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        lock = asyncio.Lock()

        async def send(message: dict) -> None:  # server-initiated notifications
            async with lock:
                writer.write((dumps(message) + "\n").encode())
                await writer.drain()

        session = server.create_session(transport="tcp", send=send,
                                        principal=Principal(id="tcp-client", roles={"default"}))
        try:
            while line := await reader.readline():
                try:
                    msg = json.loads(line)
                except ValueError:
                    await send(error_response(None, PARSE_ERROR, "Parse error"))
                    continue
                response = await session.handle(msg)
                if response is not None:
                    await send(response)
        finally:
            session.close()
            writer.close()

    return await asyncio.start_server(on_client, host, port)

Driving it with initialize, notifications/initialized and a tools/call of the test registry's echo tool printed:

{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-11-25","capabilities":{"tools":{"listChanged":true},"resources":{"subscribe":true,"listChanged":true},
{"jsonrpc":"2.0","id":2,"result":{"content":[{"type":"text","text":"{\n  \"text\": \"hihi\"\n}"}],"isError":false,"structuredContent":{"text":"hihi"},"_meta":{"

This version handles one request at a time per connection, so notifications/cancelled cannot reach a running request. stdio.py shows how to run requests as tasks (everything but initialize, notifications and responses) while keeping writes serialized.

A checklist for any adapter or transport:

• Resolve the principal from transport facts with resolve_principal(resolver, info); never from clientInfo.
• Never write protocol data and logs to the same stream.
• Keep progress monotonic per request.
• Hop to the event loop before touching sessions from change listeners.
• Do not cache list results across principals; on 2026 results keep cacheScope: "private".
• Add a test module that runs the shared _adapter_testkit.build_layer() through the new path: definitions identical to layer.tools_wire(), a structured call, an error call, progress, a resource read, a completion, a prompt, a change notification.