核心架构

    class Protocol<Request, Notification, Result> {
        // 处理传入请求
        setRequestHandler<T>(schema: T, handler: (request: T, extra: RequestHandlerExtra) => Promise<Result>): void
    
        // 处理传入通知
        setNotificationHandler<T>(schema: T, handler: (notification: T) => Promise<void>): void
    
        // 发送请求并等待响应
        request<T>(request: Request, schema: T, options?: RequestOptions): Promise<T>
    
        // 发送单向通知
        notification(notification: Notification): Promise<void>
    }

class Session(BaseSession[RequestT, NotificationT, ResultT]):
    async def send_request(
        self,
        request: RequestT,
        result_type: type[Result]
    ) -> Result:
        """
        Send request and wait for response. Raises McpError if response contains error.
        """
        # Request handling implementation
 
    async def send_notification(
        self,
        notification: NotificationT
    ) -> None:
        """Send one-way notification that doesn't expect response."""
        # Notification handling implementation
 
    async def _received_request(
        self,
        responder: RequestResponder[ReceiveRequestT, ResultT]
    ) -> None:
        """Handle incoming request from other side."""
        # Request handling implementation
 
    async def _received_notification(
        self,
        notification: ReceiveNotificationT
    ) -> None:
        """Handle incoming notification from other side."""
        # Notification handling implementation

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
 
const server = new Server({
    name: "example-server",
    version: "1.0.0"
}, {
    capabilities: {
    resources: {}
    }
});
 
// 处理请求
server.setRequestHandler(ListResourcesRequestSchema, async () => {
    return {
    resources: [
        {
        uri: "example://resource",
        name: "示例资源"
        }
    ]
    };
});
 
// 连接传输
const transport = new StdioServerTransport();
await server.connect(transport);

import asyncio
import mcp.types as types
from mcp.server import Server
from mcp.server.stdio import stdio_server
 
app = Server("example-server")
 
@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="example://resource",
            name="Example Resource"
        )
    ]
 
async def main():
    async with stdio_server() as streams:
        await app.run(
            streams[0],
            streams[1],
            app.create_initialization_options()
        )
 
if __name__ == "__main__":
    asyncio.run(main)