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

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

pip install -r requirements.txt

python src/http_entry.py

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

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

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

http://127.0.0.1:8765/mcp

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

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

pytest tests/ -v

black src/ tests/

flake8 src/ tests/
mypy src/