AI代码协调器

// claude_desktop_config.json (or any MCP client)
{
  "mcpServers": {
    "mcp-coordinator": {
      "command": "uv",
      "args": [
        "--directory",
        "your-path-to/mcp_coordinator",
        "run",
        "mcp-coordinator-server"
      ],
      "env": {
        "MCP_SERVERS_CONFIG": "your-path-to/mcp_coordinator/mcp_servers.json",
        "PYTHONUNBUFFERED": "1"
      }
    }
  }
}

# [ROLE]
You are the {SPECIFIC_ROLE} (e.g., "Epistemic Architect" or "Repo Debugger").
- **Operating Protocol**: Foundation (Python) -> Direct Action (Tools) -> Delegate (Sub-Agents).
- **Critical Constraint**: Always work in `{ABSOLUTE_PATH}`. Never assume relative paths.

# [CONTEXT & STATE]
- **Project ID**: {PROJ_ID}
- **Current Phase**: {EVALUATE | PLAN | EXECUTE | VERIFY}
- **Institutional Memory**:
  > {SUMMARY_FROM_SEARCH_MEMORY}

# [CAPABILITIES]
- **Primary Engine**: `execute_code` (Python 3.11 + standard DS libs).
- **Knowledge Graph**: `chatdag` (Ingest/Recall/Crystallize).
- **Reasoning**: `advanced_reasoning` (Deep thought) & `verifier-graph`.
- **Skills**: `from skills import *` (Use specialized domain logic).

# [EXECUTION PLAN]
1. **Ingest**: Call `chatdag.ingest` on the target directory.
2. **Retrieve**: Search memory for similar past tasks.
3. **Execute**: Use `execute_code` for data processing or `call_tool` for single-shot actions.
4. **Crystallize**: Synthesize findings into a final artifact.

# [OUTPUT FORMAT]
All final deliverables must be saved to `/knowledge_base/outputs/` and summarized in JSON:
{
  "status": "success",
  "artifact_path": "...",
  "crystallized_insight": "..."
}

# [ROLE] Repository Architect
You are an expert in codebase analysis and refactoring.

# [CONTEXT]
Path: /home/ty/Repositories/my-project
Goal: Implement async-safe logging across the entire repo.

# [PLAN]
1. Use `ast-analyzer` to identify all synchronous logging calls.
2. Use `execute_code` to generate a refactoring map.
3. Use `coordinator_task` to apply changes to individual modules in parallel.
4. Verify changes with `local-repl` by running unit tests.

# [MEMORY]
Check `search_memory("logging patterns")` for preferred library usage.

# From source
git clone https://github.com/angrysky56/mcp_coordinator.git
cd mcp_coordinator

# Install dependencies
uv tool install .

# Start infrastructure (one-time or after reboot)
docker compose up -d                    # Main executor
./start_chatdag_infrastructure.sh       # ChatDAG (if using semantic memory)

# Verify containers are running
docker ps

./start.sh --gpu

uv run mcp-coordinator run "Research quantum computing and save a summary to outputs/"

Instructions:
You are an Engineer Agent. Your goal is to BUILD capability, not just answer questions.

Core Protocol:
1. CHECK: Do we have a skill for this? (search_skills)
2. BUILD: If not, write it, test it, and save it. (save_skill)
3. DELEGATE: If complex, spawn a sub-agent. (coordinator_task)

This means I can:

Find the right tool semantically: find_tools("semantic search for papers")
Call it directly: call_tool("arxiv-mcp-server", "search_papers", {...})
Write custom orchestration code that chains multiple servers
Save that orchestration as a reusable skill for later
Delegate entire research projects to sub-agents (Depth-limited recursion)
Maintain semantic memory of results across sessions

### Available Tools

**Server Discovery & Coordination**

| Tool | Description |
|------|-------------|
| `discover_servers` | Scan all configured MCP servers and cache their capabilities |
| `list_servers` | Get available server names, optionally filtered by tag |
| `find_tools` | Semantic search: "find tools for database queries" → returns matching tools |
| `list_tools` | Get all tools from a specific server |
| `get_tool_schema` | Get the full input/output schema for any tool |
| `call_tool` | Execute any tool on any connected MCP server |
| `generate_tools` | Auto-generate Python wrapper libraries for all servers |

**Code Execution**
| Tool | Description |
|------|-------------|
| `execute_code` | Run Python code in a sandboxed Docker container |
| `execute_file` | Run a Python file in the sandbox |

**Skills (Reusable Code)**
| Tool | Description |
|------|-------------|
| `save_skill` | Save Python code as a named, versioned, reusable skill |
| `list_skills` | Show all saved skills |
| `get_skill` | Retrieve a skill's code and metadata |
| `search_skills` | Find skills by name, description, or tags |
| `run_skill` | Execute a saved skill with arguments |
| `get_skill_import` | Get the import statement to use a skill in code |
| `delete_skill` | Remove a skill from the library |

**Memory (Semantic Long-Term Storage)**
| Tool | Description |
|------|-------------|
| `save_memory` | Store text in ChromaDB with optional metadata |
| `search_memory` | Semantic search across saved memories |

**Agent Delegation**
| Tool | Description |
|------|-------------|
| `coordinator_task` | Spawn an internal sub-agent to handle complex multi-step tasks |

**No setup beyond the config line.** You ask, the AI figures out which tools to use.

---

## Why This Matters

| Traditional MCP | MCP-Coordinator |
|-----------------|-----------------|
| Load 100+ tools upfront | Semantic search finds what you need |
| No code execution | Full Python in Docker sandbox |
| Results vanish after session | Writes to your filesystem |
| Manual tool registration | Skills created via API, auto-indexed |
| One AI, one context | Sub-agents handle sub-tasks |
| ~150k tokens per workflow | ~2k tokens (Anthropic's testing) |

---

## Configuration

Environment variables (`.env` or shell):

**Core Settings**
```bash
MCP_EXECUTOR_TYPE=docker          # docker | local (docker recommended)
MCP_KNOWLEDGE_BASE_DIR=./knowledge_base
MCP_WORKSPACE_DIR=./workspace
MCP_ADDITIONAL_MOUNTS=/path/to/extra/dir1,/path/to/extra/dir2 # Optional: Mount extra host directories
MCP_SERVERS_CONFIG=./mcp_servers.json

OPENROUTER_API_KEY=sk-or-...      # Required for sub-agents
OPENROUTER_MODEL=google/gemini-2.0-flash-exp:free
OPENROUTER_TRANSFORMS=true        # Auto-compress prompts > context size
MAX_AGENT_HISTORY=50              # Sliding window for agent conversation
MAX_COMPLETION_TOKENS=            # Optional hard limit on output tokens

MCP_TOOL_LOGGING=true             # Log all tool calls to SQLite (default: on)
MCP_SANITIZE_LOGS=false           # Redact API keys, CC#s, SSNs from logs

MCP_DISCOVERY_TIMEOUT=30          # Seconds to wait for server discovery
MCP_EXECUTION_TIMEOUT=300         # Seconds for code execution

# Executor (docker recommended for security)
MCP_EXECUTOR_TYPE=docker

# API Key (OpenRouter recommended)
OPENROUTER_API_KEY=sk-or-v1-xxxxx
OPENROUTER_MODEL=google/gemini-2.0-flash-exp:free

# Paths
MCP_KNOWLEDGE_BASE_DIR=./knowledge_base
MCP_WORKSPACE_DIR=./workspace

# Context management
CONTEXT_STRATEGY=transforms
OPENROUTER_TRANSFORMS=true

await execute_code("""
import pandas as pd
df = pd.read_csv('/knowledge_base/data.csv')
print(df.describe())
""")
# Runs in Docker. Results persist to your disk.

tools = await find_tools("search academic papers")
# → Finds arxiv-mcp-server without you knowing the exact name

# I can create this:
await save_skill("web_scraper", '''
def scrape(url: str) -> str:
    import requests
    return requests.get(url).text[:5000]
''')

# Then use it anytime:
await run_skill("web_scraper", {"url": "https://example.com"})

await coordinator_task(
    "Research transformer architectures and save a summary to outputs/",
    model="google/gemini-2.0-flash-exp:free"
)
# Sub-agent handles the entire workflow internally.
# Recursion depth is tracked automatically (Max Depth = 2) to prevent infinite loops.

await save_memory("User prefers concise responses")
# ...later...
context = await search_memory("user preferences")

uv run mcp-coordinator run "Research quantum computing and save a summary to outputs/"

mcp_coordinator/
├── src/mcp_coordinator/     # Core coordinator code
│   ├── agent.py             # Internal agent with entropy support
│   ├── coordinator.py       # Main orchestration logic
│   ├── server.py            # MCP tool definitions
│   └── executor.py          # Docker/local execution
├── knowledge_base/          # Persistent storage
│   ├── skills/              # Saved Python skills
│   ├── entropy_traces/      # Saved seed chains
│   ├── chroma/              # Vector database
│   └── outputs/             # Generated deliverables
├── mcp_tools/               # Auto-generated server wrappers
└── mcp_servers.json         # Server configuration

await call_tool("chatdag", "ingest_data", {"path": "/path/to/code", "recursive": True})
await call_tool("chatdag", "crystallize_thought", {"topic": "system architecture"})

await call_tool("reflective_agent_architecture", "prove", {
    "premises": ["all x (human(x) -> mortal(x))", "human(socrates)"],
    "conclusion": "mortal(socrates)"
})
await call_tool("reflective_agent_architecture", "check_cognitive_state", {})