传输机制

const server = new Server({
  name: "example-server",
  version: "1.0.0"
}, {
  capabilities: {}
});
 
const transport = new StdioServerTransport();
await server.connect(transport);

const client = new Client({
  name: "example-client",
  version: "1.0.0"
}, {
  capabilities: {}
});
 
const transport = new StdioClientTransport({
  command: "./server",
  args: ["--option", "value"]
});
await client.connect(transport);

app = Server("example-server")
 
async with stdio_server() as streams:
    await app.run(
        streams[0],
        streams[1],
        app.create_initialization_options()
    )

params = StdioServerParameters(
    command="./server",
    args=["--option", "value"]
)
 
async with stdio_client(params) as streams:
    async with ClientSession(streams[0], streams[1]) as session:
        await session.initialize()

import express from "express";
 
const app = express();
 
const server = new Server({
  name: "example-server",
  version: "1.0.0"
}, {
  capabilities: {}
});
 
let transport: SSEServerTransport | null = null;
 
app.get("/sse", (req, res) => {
  transport = new SSEServerTransport("/messages", res);
  server.connect(transport);
});
 
app.post("/messages", (req, res) => {
  if (transport) {
    transport.handlePostMessage(req, res);
  }
});
 
app.listen(3000);

const client = new Client({
  name: "example-client",
  version: "1.0.0"
}, {
  capabilities: {}
});
 
const transport = new SSEClientTransport(
  new URL("http://localhost:3000/sse")
);
await client.connect(transport);

from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route
 
app = Server("example-server")
sse = SseServerTransport("/messages")
 
async def handle_sse(scope, receive, send):
    async with sse.connect_sse(scope, receive, send) as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())
 
async def handle_messages(scope, receive, send):
    await sse.handle_post_message(scope, receive, send)
 
starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Route("/messages", endpoint=handle_messages, methods=["POST"]),
    ]
)

async with sse_client("http://localhost:8000/sse") as streams:
    async with ClientSession(streams[0], streams[1]) as session:
        await session.initialize()

interface Transport {
  // 开始处理消息
  start(): Promise<void>;
 
  // 发送JSON-RPC消息
  send(message: JSONRPCMessage): Promise<void>;
 
  // 关闭连接
  close(): Promise<void>;
 
  // 回调函数
  onclose?: () => void;
  onerror?: (error: Error) => void;
  onmessage?: (message: JSONRPCMessage) => void;
}

@contextmanager
async def create_transport(
    read_stream: MemoryObjectReceiveStream[JSONRPCMessage | Exception],
    write_stream: MemoryObjectSendStream[JSONRPCMessage]
):
    """
    Transport interface for MCP.

    Args:
        read_stream: Stream to read incoming messages from
        write_stream: Stream to write outgoing messages to
    """
    async with anyio.create_task_group() as tg:
        try:
            # Start processing messages
            tg.start_soon(lambda: process_messages(read_stream))

            # Send messages
            async with write_stream:
                yield write_stream

        except Exception as exc:
            # Handle errors
            raise exc
        finally:
            # Clean up
            tg.cancel_scope.cancel()
            await write_stream.aclose()
            await read_stream.aclose()