MOP

MOPは、複数のAIエージェント（LLM）が協調してタスクを遂行するためのMCPサーバーです。OpenCodeとCodexの両方で利用可能な汎用MCPサーバーとして設計し、エージェント間の役割分担・通信・成果物の統合を自動化します。

アーキテクチャ

User Request
    │
    ▼
┌─────────────────────────────────────┐
│         MCP Orchestrator Server      │
│  (Python FastMCP / TypeScript SDK)   │
│                                      │
│  tools:                              │
│    decompose_task                    │
│    assign_agent                      │
│    agent_communicate                 │
│    get_task_status                   │
│    submit_artifact                   │
│    review_code                       │
│    merge_results                     │
│                                      │
│  resources:                          │
│    task://{task_id}/status           │
│    agent://{agent_id}/inbox          │
│    artifact://{artifact_id}          │
└─────────────────────────────────────┘
    │
    ▼
 Clients register themselves with register_agent

エージェント構成

MOP は起動時にエージェントを登録しません。list_registered_agents は、まだ register_agent を呼んだクライアントがなければ空の配列を返します。各クライアントは 共有 HTTP endpoint へ接続した後、安定した ID で自身を登録してください。再接続時の 同じ ID による登録は、モデル・名前・役割を最新の値へ更新します。

セットアップ

1. 仮想環境の作成

python3 -m venv .venv
source .venv/bin/activate

2. 依存関係インストール

pip install -r requirements.txt

3. 共有HTTP MCPサーバー起動

python src/http_entry.py

http://127.0.0.1:8765/mcp でStreamable HTTP MCPサーバーが待機します。 Codex、Kimiなど、通信させたい全クライアントをこの同じURLへ接続してください。 タスク、成果物、メッセージはこの単一プロセス内で共有されます。

INFO Starting shared MOP server at http://127.0.0.1:8765/mcp
INFO Starting MCP server 'mop' with transport 'streamable-http'

ホスト、ポート、パスはオプションまたは環境変数で変更できます。

python src/http_entry.py --host 127.0.0.1 --port 9000 --path /mcp
MOP_PORT=9000 python src/http_entry.py

4. Codex設定

プロジェクトの .codex/config.toml:

[mcp_servers.mop]
url = "http://127.0.0.1:8765/mcp"
enabled = true
tool_timeout_sec = 60
default_tools_approval_mode = "auto"

5. OpenCode 設定

~/.config/opencode/opencode.json:

{
  "mcp": {
    "mop": {
      "type": "remote",
      "url": "http://127.0.0.1:8765/mcp",
      "enabled": true
    }
  }
}

6. 他クライアントの設定

クライアントのMCP設定で、トランスポートに Streamable HTTP、URLに次を指定します。

http://127.0.0.1:8765/mcp

クライアントごとに src/mcp_entry.py を起動すると、stdioプロセスごとに状態が分離し、 エージェント間のメッセージは共有されません。stdio互換モードは単独利用の場合に限り、 次のコマンドで起動できます。

python src/mcp_entry.py

プロジェクト構成

rade/
├── docs/
│   └── 001-multi-agent-mcp-orchestrator.md
├── src/
│   ├── orchestrator/
│   │   ├── __init__.py
│   │   ├── server.py           # MCPサーバーエントリポイント
│   │   ├── tools/
│   │   │   ├── __init__.py
│   │   │   ├── decompose.py    # decompose_taskツール
│   │   │   ├── assign.py       # assign_agentツール
│   │   │   ├── communicate.py  # agent_communicateツール
│   │   │   ├── status.py       # get_task_statusツール
│   │   │   ├── artifact.py     # submit_artifactツール
│   │   │   ├── review.py       # review_codeツール
│   │   │   └── merge.py        # merge_resultsツール
│   │   ├── models/
│   │   │   ├── __init__.py
│   │   │   ├── task.py         # Taskモデル定義
│   │   │   ├── agent.py        # Agentモデル定義
│   │   │   └── artifact.py     # Artifactモデル定義
│   │   ├── storage/
│   │   │   ├── __init__.py
│   │   │   ├── task_store.py   # タスクストア実装
│   │   │   └── artifact_store.py # アーティファクトストア実装
│   │   └── orchestrator.py     # コアロジック
│   ├── http_entry.py           # 共有Streamable HTTPサーバー
│   └── mcp_entry.py            # 単独利用向けstdioサーバー
├── tests/
│   ├── test_decompose.py
│   ├── test_assign.py
│   ├── test_communicate.py
│   ├── test_merge.py
│   └── test_orchestrator.py
├── pyproject.toml
├── requirements.txt
└── README.md

実装担当

Agent A — Kimi K2.7 Code（Planner）

• src/orchestrator/tools/decompose.py — decompose_task の全実装
• src/orchestrator/tools/assign.py — assign_agent の全実装
• src/orchestrator/tools/merge.py — merge_results の全実装
• src/orchestrator/orchestrator.py — コアロジック（タスク管理、状態遷移、進行管理）
• src/orchestrator/models/task.py — Taskモデル定義
• src/orchestrator/models/agent.py — Agentモデル定義
• src/orchestrator/storage/task_store.py — タスクストア実装

Agent B — Qwen3.7 Plus（Coder）

• src/orchestrator/tools/communicate.py — agent_communicate の全実装
• src/orchestrator/tools/status.py — get_task_status の全実装
• src/orchestrator/tools/artifact.py — submit_artifact の全実装
• src/orchestrator/models/artifact.py — Artifactモデル定義
• src/orchestrator/storage/artifact_store.py — アーティファクトストア実装
• src/orchestrator/server.py — MCPサーバーエントリポイント（全ツールの登録）
• src/mcp_entry.py — サーバー起動用エントリポイント

Agent C — Mimo v2.5 Pro（Tester）

• src/orchestrator/tools/review.py — review_code の全実装
• tests/test_decompose.py — decompose_task のテスト
• tests/test_assign.py — assign_agent のテスト
• tests/test_communicate.py — agent_communicate のテスト
• tests/test_merge.py — merge_results のテスト
• tests/test_orchestrator.py — 統合テスト
• pyproject.toml — プロジェクト設定（テストフレームワーク、依存関係）
• requirements.txt — Python依存パッケージ一覧
• README.md — プロジェクト概要と実行方法

MCPツール定義

decompose_task

• 説明: ユーザーリクエストをサブタスクに分解する
• パラメーター: request: str — ユーザーからの要求
• 戻り値: Task（分解結果のルートタスク、subtasksに子タスクIDリスト）

assign_agent

• 説明: 特定のタスクをエージェントに割り当てる
• パラメーター: task_id: str, agent_id: str
• 戻り値: 更新後のTask

agent_communicate

• 説明: エージェント間でメッセージを送受信する
• パラメーター: from_agent: str, to_agent: str, message_type: str, content: str, task_id: str | None
• 戻り値: Message

get_task_status

• 説明: タスクの状態を取得する
• パラメーター: task_id: str
• 戻り値: Task（ステータス、進捗、成果物一覧含む）

submit_artifact

• 説明: エージェントが成果物（コード・テスト・ドキュメント）を提出する
• パラメーター: task_id: str, agent_id: str, file_path: str, content: str, type: str
• 戻り値: Artifact

review_code

• 説明: 提出された成果物に対してレビューを実施する
• パラメーター: artifact_id: str, reviewer_agent_id: str
• 戻り値: Review（コメント、承認/却下、修正提案）

merge_results

• 説明: サブタスクの成果物を統合する
• パラメーター: task_id: str
• 戻り値: 統合後のTask（artifactsにマージ結果含む）

使用例

Pythonコードから直接利用する例:

import asyncio
from orchestrator.orchestrator import Orchestrator
from orchestrator.storage.artifact_store import ArtifactStore

async def main():
    store = ArtifactStore()
    orch = Orchestrator(artifact_store=store)

    # タスクを分解
    task = await orch.decompose_task('Build a calculator API')
    print(f'Subtasks: {task.subtasks}')

    # エージェントを割り当て
    await orch.assign_agent(task.subtasks[0], 'B-Qwen')

    # エージェント間で通信
    await orch.agent_communicate(
        'B-Qwen', 'A-Kimi', 'question',
        'What auth method?', task.subtasks[0]
    )

    # ステータス確認
    status = await orch.get_task_status(task.subtasks[0])
    print(f'Status: {status.status.value}')

asyncio.run(main())

全ワークフロー例

import asyncio
from orchestrator.orchestrator import Orchestrator
from orchestrator.storage.artifact_store import ArtifactStore
from orchestrator.tools.artifact import create_submit_artifact
from orchestrator.tools.review import ReviewTool
from orchestrator.models.task import TaskStatus

async def full_workflow():
    store = ArtifactStore()
    orch = Orchestrator(artifact_store=store)

    # 1. タスク分解
    task = await orch.decompose_task('Build a REST API')

    # 2. エージェント割り当て
    for i, sid in enumerate(task.subtasks):
        agents = ['A-Kimi', 'B-Qwen', 'B-Qwen', 'C-Mimo']
        await orch.assign_agent(sid, agents[i % 4])

    # 3. 成果物提出
    submit = create_submit_artifact(store, orch)
    for sid in task.subtasks:
        await submit(
            task_id=sid, agent_id='B-Qwen',
            file_path=f'src/{sid}.py',
            content=f'# Code for {sid}',
            type='code'
        )

    # 4. コードレビュー
    review_tool = ReviewTool(store)
    artifacts = store.list_all()
    review = await review_tool.review_code(
        artifacts[0].id, 'C-Mimo'
    )
    print(f'Review: {review.status.value}')

    # 5. サブタスク完了
    for sid in task.subtasks:
        orch.update_task_status(sid, TaskStatus.COMPLETED)

    # 6. 成果物統合
    merged = await orch.merge_results(task.id)
    print(f'Merged: {merged.status.value}, {len(merged.artifacts)} artifacts')

asyncio.run(full_workflow())

MCPツール一覧

ツール	説明	担当エージェント
decompose_task	リクエストをサブタスクに分解	Agent A
assign_agent	タスクをエージェントに割り当て	Agent A
agent_communicate	エージェント間メッセージ送受信	Agent B
get_agent_inbox	受信箱をカーソル付きで取得	Agent B
wait_for_agent_message	新着メッセージを待機	Agent B
get_task_status	タスクステータス取得	Agent B
submit_artifact	成果物提出	Agent B
review_code	コードレビュー実施	Agent C
merge_results	成果物統合	Agent A
get_server_info	サーバーと登録ツールの診断情報	Codex

ライブ通信

受信側はまず get_agent_inbox を呼び、返された latest_message_id を保持します。 続けて wait_for_agent_message の after_message_id にそのIDを指定すると、 新着メッセージが届くまで待機できます。送信側は従来どおり agent_communicate を呼びます。

この待機はメッセージ配送をリアルタイム化しますが、停止中のLLMを自動起動するものでは ありません。受信クライアントが待機ツールを実行中である必要があります。また、現在の 状態はHTTPサーバープロセス内に保持され、サーバー再起動時に失われます。

MCPリソース

リソース	説明
task://{task_id}/status	タスクのステータス情報を取得
agent://{agent_id}/inbox	エージェントの受信箱メッセージ一覧
artifact://{artifact_id}	成果物の内容を取得

開発

テスト実行

pytest tests/ -v

コードフォーマット

black src/ tests/

静的解析

flake8 src/ tests/
mypy src/

ライセンス

MIT License