Agent Runtime MCP

代理运行时MCP

通过God-Agent集成实现跨会话AGI自治的持久任务队列和目标分解。

描述

Agent Runtime MCP提供跨会话的持久任务管理，实现真正的自主AGI工作流。功能包括：

• 持续的目标和任务：重新启动后仍能存活的SQLite支持的存储
• AI驱动的目标分解：将复杂的目标分解为可执行的任务
• 依赖管理：基于依赖关系的自动任务排序
• 优先队列：按优先级和准备状态进行智能任务调度
• 接力赛协议 （上帝代理阶段2）：48个代理管道，具有结构化交接
• 断路器 （上帝代理人阶段5）：具有自动回退功能的容错
• 跨会话连续性：在你中断的地方继续工作

安装

使用pip

git clone https://github.com/marc-shade/agent-runtime-mcp
cd agent-runtime-mcp
pip install -r requirements.txt

使用紫外线（推荐）

git clone https://github.com/marc-shade/agent-runtime-mcp
cd agent-runtime-mcp
uv pip install -r requirements.txt

依赖项

pip install anthropic-mcp

配置

添加 ~/.claude.json:

{
  "mcpServers": {
    "agent-runtime": {
      "command": "python3",
      "args": [
        "/absolute/path/to/agent-runtime-mcp/server.py"
      ]
    }
  }
}

工具

核心目标与任务管理（9）

接力赛协议（上帝代理人第二阶段）（6）

断路器（上帝代理人第五阶段：小舞者）（7）

使用示例

基本目标创建

# Create goal
goal = mcp__agent-runtime__create_goal({
    "name": "Build REST API",
    "description": "Create RESTful API for user authentication with JWT tokens",
    "metadata": {"priority": "high", "project": "auth-service"}
})
# Returns: {"id": 1, "name": "Build REST API", "status": "active", ...}

AI目标分解

# Decompose goal into tasks (sequential strategy)
result = mcp__agent-runtime__decompose_goal({
    "goal_id": 1,
    "strategy": "sequential"
})
# Returns: {
#   "goal_id": 1,
#   "strategy": "sequential",
#   "tasks_created": [101, 102, 103, 104, 105],
#   "count": 5
# }
# Tasks: Research → Plan → Implement → Test → Document (with dependencies)

并行分解

# Decompose for parallel execution
result = mcp__agent-runtime__decompose_goal({
    "goal_id": 1,
    "strategy": "parallel"
})
# Creates: Backend, Frontend, Testing tasks (no dependencies, run simultaneously)

层次分解

# Decompose into phases
result = mcp__agent-runtime__decompose_goal({
    "goal_id": 1,
    "strategy": "hierarchical"
})
# Creates: Phase 1 (Foundation) → Phase 2 (Core) → Phase 3 (Integration) → Phase 4 (Optimization)

任务队列处理

# Get next ready task
task = mcp__agent-runtime__get_next_task()
# Returns: Highest priority task with all dependencies met
# {"id": 101, "title": "Research requirements...", "priority": 10, ...}

# Start work
mcp__agent-runtime__update_task_status({
    "task_id": 101,
    "status": "in_progress"
})

# Complete task
mcp__agent-runtime__update_task_status({
    "task_id": 101,
    "status": "completed",
    "result": "Requirements documented in docs/api-spec.md"
})

# Get next (automatically handles dependencies)
next_task = mcp__agent-runtime__get_next_task()
# Returns: Task 102 (Plan approach) since Research (101) is complete

使用依赖关系手动创建任务

# Create task with explicit dependencies
mcp__agent-runtime__create_task({
    "goal_id": 1,
    "title": "Deploy to production",
    "description": "Deploy authentication service",
    "priority": 7,
    "dependencies": [103, 104]  # Wait for Implementation and Testing
})

接力赛管道（48代理商）

# Create relay pipeline for complex workflow
pipeline = mcp__agent-runtime__create_relay_pipeline({
    "name": "Research Paper Analysis",
    "goal": "Extract insights from 10 AGI papers",
    "agent_types": [
        "researcher",      # Gather papers
        "analyzer",        # Extract key points
        "synthesizer",     # Find patterns
        "validator",       # Check quality
        "formatter"        # Create report
    ],
    "token_budget": 100000
})
# Returns: {"pipeline_id": "rp_abc123", "agent_count": 5, ...}

# Check pipeline status
status = mcp__agent-runtime__get_relay_status({
    "pipeline_id": "rp_abc123"
})
# Returns: {
#   "current_step": 2,
#   "total_steps": 5,
#   "status": "in_progress",
#   "quality_scores": [0.92, 0.88, ...],
#   "tokens_used": 24531
# }

# Get current baton (context for next agent)
baton = mcp__agent-runtime__get_relay_baton({
    "pipeline_id": "rp_abc123"
})
# Returns: {
#   "baton": {...},
#   "prompt": "You are the Synthesizer. Previous output: ..."
# }

# Advance to next step
mcp__agent-runtime__advance_relay({
    "pipeline_id": "rp_abc123",
    "quality_score": 0.88,
    "l_score": 0.85,
    "output_entity_id": 456,
    "tokens_used": 8234,
    "output_summary": "Found 3 key patterns across papers"
})

# Retry failed step
mcp__agent-runtime__retry_relay_step({
    "pipeline_id": "rp_abc123",
    "step_index": 2
})

断路器（容错）

# Check agent circuit breaker status
status = mcp__agent-runtime__circuit_breaker_status({
    "agent_id": "researcher_agent"
})
# Returns: {
#   "agent_id": "researcher_agent",
#   "state": "CLOSED",
#   "failure_count": 0,
#   "success_count": 42
# }

# Record failure
mcp__agent-runtime__circuit_breaker_record_failure({
    "agent_id": "researcher_agent",
    "failure_type": "timeout",
    "error_message": "API request timed out after 30s"
})

# List all circuit breakers
breakers = mcp__agent-runtime__circuit_breaker_list()
# Returns: {
#   "total_breakers": 10,
#   "open_circuits": ["failing_agent_1", "failing_agent_2"],
#   "half_open_circuits": ["recovering_agent"],
#   "breakers": [...]
# }

# Configure thresholds
mcp__agent-runtime__circuit_breaker_configure({
    "agent_id": "researcher_agent",
    "failure_threshold": 5,
    "window_seconds": 60,
    "cooldown_seconds": 300,
    "fallback_agent": "generalist"
})

# Manually trip (emergency stop)
mcp__agent-runtime__circuit_breaker_trip({
    "agent_id": "researcher_agent",
    "reason": "Manual intervention - debugging required"
})

# Reset after fix
mcp__agent-runtime__circuit_breaker_reset({
    "agent_id": "researcher_agent"
})

跨会话简历

# Session 1: Create goal and start work
goal = mcp__agent-runtime__create_goal({"name": "Big Project", ...})
mcp__agent-runtime__decompose_goal({"goal_id": goal["id"]})
task1 = mcp__agent-runtime__get_next_task()
mcp__agent-runtime__update_task_status({"task_id": task1["id"], "status": "in_progress"})

# [Close Claude Code, restart later]

# Session 2: Resume exactly where left off
pending = mcp__agent-runtime__list_tasks({"status": "in_progress"})
# Returns: [task1] - still marked as in_progress
task1_updated = mcp__agent-runtime__update_task_status({
    "task_id": task1["id"],
    "status": "completed"
})
next_task = mcp__agent-runtime__get_next_task()
# Automatically gets task2 (next in dependency chain)

需求

• python: 3.10+
• 依赖项: anthropic-mcp （MCP-SDK）
• 存储: ~/.claude/agent_runtime.db （SQLite）

数据库模式

表在 ~/.claude/agent_runtime.db:

• goals -具有状态和元数据的高级目标
• tasks -具有依赖关系、优先级和结果的单个任务
• task_queue -队列位置和调度信息
• relay_pipelines -接力赛管道定义（上帝代理人第二阶段）
• relay_batons -管道台阶的巴吞邦
• circuit_breakers -断路器状态和历史（上帝代理人第5阶段）

分解策略

顺序的

Task 1 → Task 2 → Task 3 → Task 4 → Task 5

每项任务都取决于前一项任务。线性执行。

并行

Task 1 (Backend)  ─┐
Task 2 (Frontend) ─┼─→ All run simultaneously
Task 3 (Testing)  ─┘

没有依赖关系。最大并行度。

分层的

Phase 1 (Foundation)
  ↓
Phase 2 (Core Implementation)
  ↓
Phase 3 (Integration)
  ↓
Phase 4 (Optimization)

可以进一步分解的大相。

测试

# Run test suite
python3 test_agent_runtime.py

# Test relay protocol
python3 test_relay_protocol.py

# Test circuit breaker
python3 test_circuit_breaker.py

上帝代理人整合

第二阶段：接力赛协议

• 48个代理顺序管道
• 有组织地传递接力棒
• 每一步都有质量门
• L-Score输出质量跟踪
• 单步重试（不完全重新启动）

第五阶段：断路器（小舞者）

• 自动故障检测
• 状态机：关闭→ OPEN → 半开→ 关闭
• 可配置的阈值和冷却
• 回退代理路由
• 恢复监控

链接

• github：https://github.com/marc-shade/agent-runtime-mcp
• 问题：https://github.com/marc-shade/agent-runtime-mcp/issues