Serial-MCP

Serial-MCP

エージェントがコンピューターのシリアルポートに接続されたデバイスと通信できるMCPサーバーです。現時点ではMACでのみテストされています。

FastMCPフレームワークで構築された堅牢なシリアル通信サーバーです。メッセージのバッファリング、エラー処理、接続管理などの機能を備え、シリアルポート通信のための信頼性の高いインターフェースを提供します。

インストール

1. リポジトリをクローンします。

git clone https://github.com/PaDev1/Serial-MCP.git
cd Serial-MCP

2. 依存関係をインストールします。

pip install -r requirements.txt

インストール後、次の構成でサーバーをClaude Desktopなどに追加します。

"serial_MCP": { "command": "uv", "args": [ "run", "--with", "fastmcp,pyserial", "fastmcp", "run", "/serial_MCP.py" ], "env": { "SERIAL_PORT": "", "SERIAL_BAUD_RATE": "9600", "SERIAL_BUFFER_LENGTH": "100" } }

機能

• 非同期シリアル通信：効率的なI/O操作を実現するためasyncioを使用しています。
• メッセージのバッファリング：タイムスタンプに対応した設定可能なバッファーサイズ
• 接続管理：ポートの自動検出と接続管理
• エラー処理：包括的なエラー検出と報告
• 環境設定：環境変数と実行時設定に対応
• ログ記録：タイムスタンプとエラー追跡を含む詳細なログ記録

利用可能なツール

1. delay

指定した秒数だけ待機します。

パラメーター：

• delay（float）：待機する秒数（正の値である必要があります）

戻り値：

• 成功ステータス
• 指定された待機時間
• 実際の待機時間（計測値）
• ステータスメッセージ

例：

await delay({
    "delay": 2.5  # Wait for 2.5 seconds
})

2. init_serial

指定したパラメーターでシリアル接続を初期化します。

パラメーター：

• port（str）：シリアルポートのデバイスパス（例：'/dev/tty.usbmodem1101'）
• baudrate（int）：通信速度（ボー単位、デフォルト：9600）
• buffer_length（int）：バッファーに格納するメッセージの最大数（デフォルト：100）

戻り値：

• 初期化のステータス
• 現在のモード（実接続／切断）
• ポートとボーレートの情報

3. send_message

シリアル接続を通じてメッセージを送信し、必要に応じて応答を待ちます。

パラメーター：

• message（str）：送信するメッセージ
• wait_for_response（bool）：送信後に応答を待つかどうか（デフォルト：False）
• response_timeout（float）：応答を待つ秒数（デフォルト：0.5）

戻り値：

• success（bool）：操作全体の成功状態
• status（str）：操作のステータス（「success」、「error」など）
• mode（str）：現在の接続モード（「real」、「disconnected」）
• write_status（str）：書き込み操作のステータス：
  • "success"：メッセージを正常に書き込みました
  • "failed"：書き込み操作に失敗しました
  • "timeout"：書き込み操作がタイムアウトしました
• connection_state（str）：現在の接続状態：
  • "connected"：シリアルポートは接続され、利用可能です
  • "disconnected"：アクティブなシリアル接続はありません
  • "error"：接続がエラー状態です
• response_status（str）：応答のステータス：
  • "not_checked"：応答の確認は要求されていません
  • "received"：応答を受信しました
  • "no_response"：タイムアウトまでに応答を受信しませんでした
  • "error"：応答の確認中にエラーが発生しました
• response_messages（list）：受信したメッセージのリスト（存在する場合）
• bytes_written（int）：正常に書き込まれたバイト数
• error_type（str）：操作に失敗した場合のエラーの種類
• message（str）：人が読める形式のステータスメッセージ
• error_details（dict）：該当する場合の追加のエラー情報

応答確認なしで送信する例：

await send_message({
    "message": "Hello",
    "wait_for_response": False
})
# Returns:
{
    "success": True,
    "status": "success",
    "mode": "real",
    "write_status": "success",
    "connection_state": "connected",
    "response_status": "not_checked",
    "response_messages": [],
    "bytes_written": 6,
    "message": "Message sent successfully in real mode"
}

応答確認ありで送信する例：

await send_message({
    "message": "Hello",
    "wait_for_response": True,
    "response_timeout": 1.0
})
# Returns:
{
    "success": True,
    "status": "success",
    "mode": "real",
    "write_status": "success",
    "connection_state": "connected",
    "response_status": "received",
    "response_messages": [
        {
            "timestamp": "2024-03-14T15:30:45.123456",
            "message": "Hello received"
        }
    ],
    "bytes_written": 6,
    "message": "Message sent successfully in real mode"
}

タイムアウトまでに応答がない場合の例：

await send_message({
    "message": "Hello",
    "wait_for_response": True,
    "response_timeout": 0.5
})
# Returns:
{
    "success": True,
    "status": "success",
    "mode": "real",
    "write_status": "success",
    "connection_state": "connected",
    "response_status": "no_response",
    "response_messages": [],
    "bytes_written": 6,
    "message": "Message sent successfully in real mode"
}

4. read_message

バッファーからメッセージを読み取ります。

パラメーター：

• wait（bool）：バッファーが空の場合にメッセージを待つかどうか（デフォルト：False）
• timeout（float）：メッセージを待つ時間（秒、デフォルト：1.0）

戻り値：

• タイムスタンプ付きのメッセージのリスト
• 現在のモード
• 読み取り操作のステータス

5. list_serial_ports

システムで利用可能なすべてのシリアルポートを一覧表示します。

パラメーター： なし

戻り値：

• 利用可能なポートの詳細リスト
• ポート名と説明
• ハードウェア ID

6. get_serial_status

シリアル接続の現在のステータスを取得します。

パラメーター： なし

戻り値：

• 現在のモード
• ポートとボーレートの情報
• バッファーの長さ
• 利用可能なポート
• 最後のエラーメッセージ
• シリアルモジュールの利用可否

7. configure_serial

初期化後にシリアル接続を設定します。

パラメーター：

• port（str、任意）：使用する新しいポート
• baudrate（int、任意）：使用する新しいボーレート
• list_ports（bool）：ポートの一覧表示のみを行うかどうか（デフォルト：False）

戻り値：

• 更新された設定
• 現在のモード
• 利用可能なポート（list_ports が True の場合）

8. close_serial

現在のシリアル接続を閉じます。

パラメーター： なし

戻り値：

• 切断のステータス
• 現在のモード

9. help

Serial MCP サーバーの使用方法に関する詳細な手順を取得します。

パラメーター： なし

戻り値：

• 以下を含む詳細なヘルプ情報：
  • ツールの説明とパラメーター
  • 各ツールの使用例
  • 一般的なレスポンス形式
  • 接続状態
  • レスポンスのステータス
  • 重要な使用上の注意

例：

await help()
# Returns comprehensive help information about all available tools and their usage

設定

サーバーは環境変数を使って設定できます。

• SERIAL_PORT：シリアルポートのデバイスパス（デフォルト：'/dev/tty.usb*'）
• SERIAL_BAUD_RATE：通信速度（ボー）（デフォルト：9600）
• SERIAL_BUFFER_LENGTH：バッファーに保存するメッセージの最大数（デフォルト：100）

使用方法

サーバーの起動

python serial_MCP.py

API エンドポイント

シリアル接続の初期化

await init_serial({
    "port": "/dev/tty.usbmodem1101",
    "baudrate": 9600,
    "buffer_length": 100
})

メッセージの送信

await send_message({
    "message": "Hello, device!"
})

メッセージの読み取り

await read_message({
    "wait": False,  # Whether to wait for messages
    "timeout": 1.0  # Timeout in seconds
})

利用可能なポートの一覧表示

await list_serial_ports()

接続ステータスの取得

await get_serial_status()

接続の設定

await configure_serial({
    "port": "/dev/tty.usbmodem1101",
    "baudrate": 9600,
    "list_ports": False
})

接続の切断

await close_serial()

メッセージ形式

バッファー内のメッセージはタイムスタンプ付きで保存されます。

{
    "timestamp": "2024-03-14T15:30:45.123456",  # ISO format
    "message": "Actual message content"
}

エラー処理

サーバーは、さまざまな状況に応じた詳細なエラーメッセージを提供します。

• 接続の失敗
• 無効なパラメーター
• タイムアウト
• バッファーのオーバーフロー
• 無効なメッセージ

ロギング

ログは logs/serial_MCP.log に次の形式で書き込まれます。

[timestamp] - [level] - [message]

要件

• Python 3.7+
• pyserial>=3.5
• fastmcp>=0.1.0
• pydantic>=2.0.0
• asyncio>=3.4.3

コントリビューション

1. リポジトリをフォークします
2. フィーチャーブランチを作成します（git checkout -b feature/amazing-feature）
3. 変更をコミットします（git commit -m 'Add some amazing feature'）
4. ブランチにプッシュします（git push origin feature/amazing-feature）
5. プルリクエストを作成します

ライセンス

このプロジェクトは MIT ライセンスの下でライセンスされています。詳細については LICENSE ファイルをご覧ください。

謝辞

• サーバー基盤を提供する FastMCP フレームワーク
• シリアル通信機能を提供する pyserial