Core architecture

The Model Context Protocol (MCP) is built on a flexible, extensible architecture that enables seamless communication between LLM applications and integrations. This document covers the core architectural components and concepts.

Overview

MCP follows a client-server architecture where:

Hosts are LLM applications (like Claude Desktop or IDEs) that initiate connections
Clients maintain 1:1 connections with servers, inside the host application
Servers provide context, tools, and prompts to clients

flowchart LR
    subgraph " Host (e.g., Claude Desktop) "
        client1[MCP Client]
        client2[MCP Client]
    end
    subgraph "Server Process"
        server1[MCP Server]
    end
    subgraph "Server Process"
        server2[MCP Server]
    end

    client1 <-->|Transport Layer| server1
    client2 <-->|Transport Layer| server2

Core components

Protocol layer

The protocol layer handles message framing, request/response linking, and high-level communication patterns.

    class Protocol<Request, Notification, Result> {
        // Handle incoming requests
        setRequestHandler<T>(schema: T, handler: (request: T, extra: RequestHandlerExtra) => Promise<Result>): void

        // Handle incoming notifications
        setNotificationHandler<T>(schema: T, handler: (notification: T) => Promise<void>): void

        // Send requests and await responses
        request<T>(request: Request, schema: T, options?: RequestOptions): Promise<T>

        // Send one-way notifications
        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

Key classes include:

Protocol
Client
Server

Transport layer

The transport layer handles the actual communication between clients and servers. MCP supports multiple transport mechanisms:

Stdio transport
- Uses standard input/output for communication
- Ideal for local processes
HTTP with SSE transport
- Uses Server-Sent Events for server-to-client messages
- HTTP POST for client-to-server messages

All transports use JSON-RPC 2.0 to exchange messages. See the specification for detailed information about the Model Context Protocol message format.

Message types

MCP has these main types of messages:

Requests expect a response from the other side:

interface Request {
  method: string;
  params?: { ... };
}

Notifications are one-way messages that don’t expect a response:

interface Notification {
  method: string;
  params?: { ... };
}

Results are successful responses to requests:

interface Result {
  [key: string]: unknown;
}

Errors indicate that a request failed:

interface Error {
  code: number;
  message: string;
  data?: unknown;
}

Connection lifecycle

1. Initialization

sequenceDiagram
    participant Client
    participant Server

    Client->>Server: initialize request
    Server->>Client: initialize response
    Client->>Server: initialized notification

    Note over Client,Server: Connection ready for use

Client sends initialize request with protocol version and capabilities
Server responds with its protocol version and capabilities
Client sends initialized notification as acknowledgment
Normal message exchange begins

2. Message exchange

After initialization, the following patterns are supported:

Request-Response: Client or server sends requests, the other responds
Notifications: Either party sends one-way messages

3. Termination

Either party can terminate the connection:

Clean shutdown via close()
Transport disconnection
Error conditions

Error handling

MCP defines these standard error codes:

enum ErrorCode {
  // Standard JSON-RPC error codes
  ParseError = -32700,
  InvalidRequest = -32600,
  MethodNotFound = -32601,
  InvalidParams = -32602,
  InternalError = -32603
}

SDKs and applications can define their own error codes above -32000.

Errors are propagated through:

Error responses to requests
Error events on transports
Protocol-level error handlers

Implementation example

Here’s a basic example of implementing an MCP server:

    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: {}
      }
    });

    // Handle requests
    server.setRequestHandler(ListResourcesRequestSchema, async () => {
      return {
        resources: [
          {
            uri: "example://resource",
            name: "Example Resource"
          }
        ]
      };
    });

    // Connect transport
    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)

Best practices

Transport selection

Local communication
- Use stdio transport for local processes
- Efficient for same-machine communication
- Simple process management
Remote communication
- Use SSE for scenarios requiring HTTP compatibility
- Consider security implications including authentication and authorization

Message handling

Request processing
- Validate inputs thoroughly
- Use type-safe schemas
- Handle errors gracefully
- Implement timeouts
Progress reporting
- Use progress tokens for long operations
- Report progress incrementally
- Include total progress when known
Error management
- Use appropriate error codes
- Include helpful error messages
- Clean up resources on errors

Security considerations

Transport security
- Use TLS for remote connections
- Validate connection origins
- Implement authentication when needed
Message validation
- Validate all incoming messages
- Sanitize inputs
- Check message size limits
- Verify JSON-RPC format
Resource protection
- Implement access controls
- Validate resource paths
- Monitor resource usage
- Rate limit requests
Error handling
- Don’t leak sensitive information
- Log security-relevant errors
- Implement proper cleanup
- Handle DoS scenarios

Debugging and monitoring

Logging
- Log protocol events
- Track message flow
- Monitor performance
- Record errors
Diagnostics
- Implement health checks
- Monitor connection state
- Track resource usage
- Profile performance
Testing
- Test different transports
- Verify error handling
- Check edge cases
- Load test servers

Prompts