# トランスポート

MCPの通信メカニズムについて学びます。

トランスポートは、Model Context Protocol (MCP) におけるクライアントとサーバー間の通信基盤を提供します。トランスポートは、メッセージの送受信に関する基本的なメカニズムを処理します。

## [メッセージ形式](https://modelcontextprotocol.io/docs/concepts/transports#message-format)

MCPは、ワイヤ形式として[JSON-RPC](https://www.jsonrpc.org/) 2.0を使用します。トランスポート層は、MCPプロトコルメッセージをJSON-RPC形式に変換して送信し、受信したJSON-RPCメッセージをMCPプロトコルメッセージに変換する役割を担います。

使用されるJSON-RPCメッセージには、次の3つのタイプがあります。

### [リクエスト](https://modelcontextprotocol.io/docs/concepts/transports#requests)

```typescript
{
  jsonrpc: "2.0",
  id: number | string,
  method: string,
  params?: object
}
```

### [レスポンス](https://modelcontextprotocol.io/docs/concepts/transports#responses)

```typescript
{
  jsonrpc: "2.0",
  id: number | string,
  result?: object,
  error?: {
    code: number,
    message: string,
    data?: unknown
  }
}
```

### [通知](https://modelcontextprotocol.io/docs/concepts/transports#notifications)

```typescript
{
  jsonrpc: "2.0",
  method: string,
  params?: object
}
```

## [組み込みトランスポートタイプ](https://modelcontextprotocol.io/docs/concepts/transports#built-in-transport-types)

MCPには、2つの標準トランスポート実装が含まれています。

### [標準入出力 (stdio)](https://modelcontextprotocol.io/docs/concepts/transports#standard-input%2Foutput-stdio)

stdioトランスポートは、標準入出力ストリームを介した通信を可能にします。これは、ローカル統合やコマンドラインツールに特に役立ちます。

stdioを使用する場合：

- コマンドラインツールを構築する場合
- ローカル統合を実装する場合
- シンプルなプロセス間通信が必要な場合
- シェルスクリプトを使用する場合

- TypeScript (Server)

```typescript
const server = new Server({
  name: "example-server",
  version: "1.0.0"
}, {
  capabilities: {}
});

const transport = new StdioServerTransport();
await server.connect(transport);
```

- TypeScript (Client)

```typescript
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);
```

- Python (Server)

```python
app = Server("example-server")

async with stdio_server() as streams:
    await app.run(
        streams[0],
        streams[1],
        app.create_initialization_options()
    )
```

- Python (Client)

```python
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()
```

### [Server-Sent Events (SSE)](https://modelcontextprotocol.io/docs/concepts/transports#server-sent-events-sse)

SSEトランスポートは、クライアントからサーバーへの通信にHTTP POSTリクエストを使用し、サーバーからクライアントへのストリーミングを可能にします。

SSEを使用する場合：

- サーバーからクライアントへのストリーミングのみが必要な場合
- 制限されたネットワークで作業する場合
- シンプルなアップデートを実装する場合

TypeScript (Server)

```typescript
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);
```

TypeScript (Client)

```typescript
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);
```

Python (Server)

```python
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"]),
    ]
)
```

Python (Client)

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

## [カスタムトランスポート](https://modelcontextprotocol.io/docs/concepts/transports#custom-transports)

MCPを使用すると、特定のニーズに合わせてカスタムトランスポートを簡単に実装できます。トランスポート実装は、Transportインターフェースに準拠する必要があります。

カスタムトランスポートは、次の目的で実装できます。

- カスタムネットワークプロトコル
- 特殊な通信チャネル
- 既存のシステムとの統合
- パフォーマンスの最適化

TypeScript

```typescript
interface Transport {
  // Start processing messages
  start(): Promise<void>;

  // Send a JSON-RPC message
  send(message: JSONRPCMessage): Promise<void>;

  // Close the connection
  close(): Promise<void>;

  // Callbacks
  onclose?: () => void;
  onerror?: (error: Error) => void;
  onmessage?: (message: JSONRPCMessage) => void;
}
```

Python

MCPサーバーはasyncioで実装されることが多いですが、より幅広い互換性のために、トランスポートのような低レベルのインターフェースは`anyio`で実装することをお勧めします。

```python
@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()
```

## [エラー処理](https://modelcontextprotocol.io/docs/concepts/transports#error-handling)

トランスポートの実装は、さまざまなエラーシナリオを処理する必要があります。

1. 接続エラー
2. メッセージ解析エラー
3. プロトコルエラー
4. ネットワークタイムアウト
5. リソースクリーンアップ

エラー処理の例：

- TypeScript

```typescript
class ExampleTransport implements Transport {
  async start() {
    try {
      // Connection logic
    } catch (error) {
      this.onerror?.(new Error(`Failed to connect: ${error}`));
      throw error;
    }
  }

  async send(message: JSONRPCMessage) {
    try {
      // Sending logic
    } catch (error) {
      this.onerror?.(new Error(`Failed to send message: ${error}`));
      throw error;
    }
  }
}
```

- Python

MCPサーバーはasyncioで実装されることが多いですが、より幅広い互換性のために、トランスポートのような低レベルのインターフェースは`anyio`で実装することをお勧めします。

```python
@contextmanager
async def example_transport(scope: Scope, receive: Receive, send: Send):
    try:
        # Create streams for bidirectional communication
        read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
        write_stream, write_stream_reader = anyio.create_memory_object_stream(0)

        async def message_handler():
            try:
                async with read_stream_writer:
                    # Message handling logic
                    pass
            except Exception as exc:
                logger.error(f"Failed to handle message: {exc}")
                raise exc

        async with anyio.create_task_group() as tg:
            tg.start_soon(message_handler)
            try:
                # Yield streams for communication
                yield read_stream, write_stream
            except Exception as exc:
                logger.error(f"Transport error: {exc}")
                raise exc
            finally:
                tg.cancel_scope.cancel()
                await write_stream.aclose()
                await read_stream.aclose()
    except Exception as exc:
        logger.error(f"Failed to initialize transport: {exc}")
        raise exc
```

## [ベストプラクティス](https://modelcontextprotocol.io/docs/concepts/transports#best-practices)

MCPトランスポートを実装または使用する場合：

1. 接続ライフサイクルを適切に処理する
2. 適切なエラー処理を実装する
3. 接続終了時にリソースをクリーンアップする
4. 適切なタイムアウトを使用する
5. 送信する前にメッセージを検証する
6. デバッグのためにトランスポートイベントをログに記録する
7. 必要に応じて再接続ロジックを実装する
8. メッセージキューのバックプレッシャーを処理する
9. 接続のヘルス状態を監視する
10. 適切なセキュリティ対策を実装する

## [セキュリティに関する考慮事項](https://modelcontextprotocol.io/docs/concepts/transports#security-considerations)

トランスポートを実装する場合：

### [認証と認可](https://modelcontextprotocol.io/docs/concepts/transports#authentication-and-authorization)

- 適切な認証メカニズムを実装する
- クライアントの資格情報を検証する
- 安全なトークン処理を使用する
- 認可チェックを実装する

### [データセキュリティ](https://modelcontextprotocol.io/docs/concepts/transports#data-security)

- ネットワークトランスポートにTLSを使用する
- 機密データを暗号化する
- メッセージの整合性を検証する
- メッセージサイズ制限を実装する
- 入力データをサニタイズする

### [ネットワークセキュリティ](https://modelcontextprotocol.io/docs/concepts/transports#network-security)

- レート制限を実装する
- 適切なタイムアウトを使用する
- サービス拒否シナリオを処理する
- 異常なパターンを監視する
- 適切なファイアウォールルールを実装する

## [トランスポートのデバッグ](https://modelcontextprotocol.io/docs/concepts/transports#debugging-transport)

トランスポートの問題をデバッグするためのヒント：

1. デバッグログを有効にする
2. メッセージフローを監視する
3. 接続状態を確認する
4. メッセージ形式を検証する
5. エラーシナリオをテストする
6. ネットワーク分析ツールを使用する
7. ヘルスチェックを実装する
8. リソース使用量を監視する
9. エッジケースをテストする
10. 適切なエラートラッキングを使用する