AInTandem代理MCP调度程序

使用Qwen agent SDK构建的全面的本地代理基础架构，具有 双SDK架构 （Qwen+Claude）、MCP服务器集成、任务调度、兼容OpenAI的API、Gradio GUI和双版本存储支持。

特性

🔄 双SDK架构：每个代理在Qwen代理SDK或Claude代理SDK之间进行选择

- Qwen SDK：支持多个LLM提供商（OpenAI、DeepSeek、GLM等） - Claude SDK：对计算机使用和扩展思维的原生支持 - 统一接口：两个SDK都实现了相同的功能 IAgentAdapter 接口

可定制代理：使用自定义系统提示和MCP服务器集成定义代理
MCP服务器集成：通过YAML配置无缝集成模型上下文协议服务器

- 双重运输支持：Stdio用于本地服务器，SSE用于远程/流媒体服务器 - 流媒体工具执行：通过服务器发送事件（SSE）实时流式传输响应

⚡ Websocket流媒体聊天：实时代理推理，实时分步显示

- 观察代理人的思维过程 - 颜色编码推理步骤（想法、工具使用、结果） - 连接丢失时自动重新连接

任务调度：使用Cron、Interval和一次性调度来调度自动化代理任务
OpenAI兼容的API：REST API与OpenAI的聊天完成和函数调用兼容
Gradio图形用户界面：用户友好的web界面，采用模块化选项卡架构
沙盒执行：具有资源限制和安全策略的隔离执行环境
双版存储：支持个人（SQLite）和企业（PostgreSQL+Redis）部署

快速入门：WebSocket流媒体聊天

通过3个步骤体验实时代理推理：

启动应用程序:

   python main.py

打开图形用户界面:

- 引导到http://localhost:7860 - 点击 "⚡ 实时聊天” 标签

连接和聊天:

- 点击 “连接” 建立WebSocket连接 - 选择一个代理（例如“研究人员”） - 键入您的消息并单击 “发送” - 观看实时出现的推理步骤！

你会看到什么:

💭 蓝卡——代理人的思维过程
🔧 橙色卡片-被称为工具
📊 紫色卡片-工具执行结果
✅ 绿卡-最终答案

有关详细文档，请参阅 docs/websocket-streaming-reasoning.md.

HTTP流API

为API客户端体验与OpenAI兼容的流式响应：

Python SDK使用

from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="test"  # Not used for authentication
)

stream = client.chat.completions.create(
    model="researcher",
    messages=[{"role": "user", "content": "Count to 5"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

cURL用法

curl -N http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "researcher",
    "messages": [{"role": "user", "content": "Say hello"}],
    "stream": true
  }'

LangChain集成

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    base_url="http://localhost:8000/v1",
    model="researcher",
    streaming=True
)

response = await llm.ainvoke("Say hello")

特征：

OpenAI兼容的SSE（服务器发送事件）格式
令牌级或块级流媒体
与OpenAI Python SDK兼容
适用于LangChain、LlamaIndex等

双SDK架构

系统支持 Qwen代理SDK和Claude代理SDK 同时，允许您根据每个代理的要求选择最佳的SDK。

SDK比较

功能	Qwen SDK	Claude SDK
多LLM支持	✅ OpenAI、DeepSeek、GLM等。	❌ 仅限克劳德模特
计算机使用	❌	✅ 直接浏览器/系统控制
拓展思维	❌	✅ 深度推理模式
MCP集成	✅	✅
函数调用	✅	✅ 本地支持
流媒体	✅	✅ 本地支持

何时使用每个SDK

使用Qwen SDK：

多LLM部署（成本优化）
使用非克劳德模型（GPT-4、DeepSeek、GLM）的代理
简单的查询响应任务
大量自动化任务

使用Claude SDK：

计算机使用任务（网络自动化、表单填写、数据输入）
需要扩展思维的复杂推理
受益于Claude原生工具调用的任务
需要浏览器交互功能的代理

配置

在中配置SDK选择 config/agents.yaml:

agents:
  # Qwen SDK agent (default)
  - name: "researcher"
    llm_model: "glm-4.7"
    sdk: "qwen"  # Optional - auto-detected from model name
    mcp_servers: ["filesystem", "web-search"]
    enabled: true

  # Claude SDK agent with Computer Use
  - name: "browser_assistant"
    llm_model: "claude-3-5-sonnet-20241022"
    sdk: "claude"  # Required for Claude-specific features
    computer_use_enabled: true
    extended_thinking_enabled: true
    mcp_servers: ["filesystem"]
    enabled: false  # Set to true when ANTHROPIC_API_KEY is configured

Claude SDK设置

要使用Claude SDK功能（计算机使用、扩展思维）：

安装Claude SDK:

   pip install anthropic

设置API密钥:

   export ANTHROPIC_API_KEY=your_key_here

将Claude提供者添加到 config/llm.yaml:

   providers:
     claude:
       api_key: "${ANTHROPIC_API_KEY}"
       base_url: https://api.anthropic.com/v1
       description: Anthropic Claude - Computer Use & Extended Thinking

启用Claude代理 在 config/agents.yaml

有关详细文档，请参阅 docs/DUAL_SDK_ARCHITECTURE.md.

建筑

┌─────────────────────────────────────────────────────────────────┐
│                     AInTandem Agent MCP Scheduler               │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐              │
│  │   Gradio    │  │   FastAPI   │  │     CLI     │              │
│  │     GUI     │  │     API     │  │  Interface  │              │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘              │
│         │                │                │                     │
│         └────────────────┴────────────────┘                     │
│                          │                                      │
│         ┌────────────────┴────────────────┐                     │
│         │         Core Components          │                    │
│         ├──────────────────────────────────┤                    │
│         │  AgentManager │  TaskScheduler   │                    │
│         │  MCPBridge    │  SandboxManager  │                    │
│         │  ConfigManager│  SecurityPolicy  │                    │
│         └──────────────────────────────────┘                    │
│                          │                                      │
│         ┌────────────────┴────────────────┐                     │
│         │            Agents               │                     │
│         ├─────────────────────────────────┤                     │
│         │  Researcher  │  Developer       │                     │
│         │  Writer      │  Analyst         │                     │
│         └─────────────────────────────────┘                     │
│                          │                                      │
│         ┌────────────────┴────────────────┐                     │
│         │          MCP Servers            │                     │
│         ├─────────────────────────────────┤                     │
│         │  Filesystem  │  Web-Search      │                     │
│         │  GitHub      │  PostgreSQL      │                     │
│         │  Google-Maps │  Puppeteer       │                     │
│         └─────────────────────────────────┘                     │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

快速开始

先决条件

Python 3.10+
pip或uv
Node.js（用于MCP服务器）

安装

# Clone the repository
git clone 
cd agent-infra

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

配置

复制示例配置：

cp .env.example .env

编辑 .env 使用API密钥：

# LLM Provider (OpenAI-compatible)
OPENAI_API_KEY=your_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1

# Or for DeepSeek
DEEPSEEK_API_KEY=your_deepseek_key

查看中的配置文件 config/ 目录：

- config/llm.yaml -LLM提供程序设置 - config/agents.yaml -代理定义 - config/mcp_servers.yaml -MCP服务器配置 - config/app.yaml -应用程序设置 - config/storage.yaml -存储和缓存配置

存储配置

系统支持两个具有不同存储后端的部署版本：

个人版（SQLite）

带有嵌入式SQLite数据库的单容器部署：

# config/storage.yaml
storage:
  type: sqlite
  sqlite:
    path: "./storage/data.db"
    enable_wal: true

cache:
  type: memory
  max_size: 1000
  default_ttl: 300

特性:

无外部依赖关系
轻松备份（单个文件）
适合个人使用和测试

企业版（PostgreSQL+Redis）

具有生产级存储的多实例部署：

# config/storage.yaml
storage:
  type: postgresql
  postgresql:
    host: "${DB_HOST}"
    port: 5432
    database: qwen_agent
    user: "${DB_USER}"
    password: "${DB_PASSWORD}"
    pool_size: 20

cache:
  type: redis
  redis:
    host: "${REDIS_HOST}"
    port: 6379
    password: "${REDIS_PASSWORD}"

特性:

水平缩放支持
连接池
分布式缓存
适用于生产环境和多用户环境

基于文件的存储（传统）

为了向后兼容，系统支持基于文件的存储：

storage:
  type: file
cache:
  type: none

这使用JSON文件 storage/tasks/ 为了坚持。

运行应用程序

# Start all services (API + GUI)
python main.py

# Or start individual components
python -m api.openapi_server     # API server on :8000
python -m gui.app                # GUI on :7860

接入点

Gradio图形用户界面: http://localhost:7860
API文件: http://localhost:8000/docs
健康检查: http://localhost:8000/health

配置

LLM配置(`config/llm.yaml`)

llm:
  provider: "openai_compatible"
  base_url: "https://api.openai.com/v1"
  api_key: "${OPENAI_API_KEY}"
  default_model: "gpt-4"
  models:
    - name: "gpt-4"
      provider: "openai"
    - name: "qwen-turbo"
      provider: "dashscope"

代理配置(`config/agents.yaml`)

agents:
  - name: "researcher"
    role: "Research Assistant"
    description: "Professional research assistant with web search capabilities"
    system_prompt: |
      You are a professional research assistant. Help users find and analyze information.
    mcp_servers: ["filesystem", "web-search"]
    llm_model: "gpt-4"
    enabled: true

  - name: "developer"
    role: "Code Assistant"
    description: "Programming assistant with code execution capabilities"
    system_prompt: |
      You are a programming assistant. Help users write, debug, and optimize code.
    mcp_servers: ["filesystem", "github"]
    llm_model: "gpt-4"
    enabled: true

MCP服务器配置(`config/mcp_servers.yaml`)

系统通过两种传输类型支持MCP服务器：

标准：标准输入/输出（默认值，适用于本地MCP服务器）
SSE：服务器发送事件（用于远程/流式MCP服务器）

有关MCP服务器的详细配置，请参阅 docs/MCP_SERVER_CONFIGURATION.md.

快速示例:

mcp_servers:
  # Local stdio server
  - name: "filesystem"
    transport: "stdio"
    command: "npx"
    args: ["-y", "@modelcontextprotocol/server-filesystem", "${AGENT_ROOT_PATH}"]
    enabled: true

  # Remote SSE server (streaming support)
  - name: "remote-mcp-server"
    transport: "sse"
    sse:
      url: "https://api.example.com/mcp/sse"
      headers:
        Authorization: "Bearer ${MCP_SERVER_TOKEN}"
    enabled: false

可用的官方服务器:

@modelcontextprotocol/server-filesystem -文件系统访问
@modelcontextprotocol/server-brave-search -网络搜索
mcp-server-github -GitHub集成
mcp-server-postgres -PostgreSQL数据库
@modelcontextprotocol/server-google-maps -谷歌地图服务
@modelcontextprotocol/server-puppeteer -Web自动化

API使用

聊天完成（非流媒体）

curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "researcher",
    "messages": [
      {"role": "user", "content": "Search for recent AI papers"}
    ],
    "temperature": 0.7
  }'

聊天完成（流媒体）

curl -N http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "researcher",
    "messages": [
      {"role": "user", "content": "Count to 5"}
    ],
    "stream": true
  }'

响应格式（SSE）：

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","created":1234567890,"model":"researcher","choices":[{"delta":{"content":"1"},"finish_reason":null}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","created":1234567890,"model":"researcher","choices":[{"delta":{"content":", 2"},"finish_reason":null}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","created":1234567890,"model":"researcher","choices":[{"delta":{},"finish_reason":"stop"}]}

函数调用（创建计划任务）

curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "researcher",
    "messages": [
      {"role": "user", "content": "Schedule a daily report at 9 AM"}
    ],
    "tools": [
      {
        "type": "function",
        "function": {
          "name": "create_scheduled_task",
          "arguments": "{\"name\": \"Daily Report\", \"agent_name\": \"analyst\", \"task_prompt\": \"Generate daily report\", \"schedule_type\": \"cron\", \"schedule_value\": \"0 9 * * *\", \"repeat\": true}"
        }
      }
    ]
  }'

可用端点

端点	方法	描述
`/v1/chat/completions`	POST	聊天完成（兼容OpenAI）
`/v1/agents`	GET	列出所有代理
`/v1/agents/{name}`	GET	获取代理详细信息
`/v1/tasks`	GET	列出所有任务
`/v1/tasks/{id}`	GET	获取任务详细信息
`/v1/tasks/{id}/enable`	POST	启用任务
`/v1/tasks/{id}/disable`	POST	禁用任务
`/v1/tasks/{id}`	删除	取消任务
`/sse/tools/call`	POST	流MCP工具执行（SSE）
`/sse/tools/call/by-name`	POST	流工具全名（SSE）
`/sse/tools`	GET	列出支持流媒体的工具
`/sse/servers`	GET	列出带有传输信息的MCP服务器
`/ws/chat/{session_id}`	WebSocket	具有推理功能的实时代理聊天

任务调度

计划类型

克龙：Unix cron表达式

   schedule_type = "cron"
   schedule_value = "0 9 * * *"  # Daily at 9 AM

间隔：两次运行之间的秒数

   schedule_type = "interval"
   schedule_value = "300"  # Every 5 minutes

曾经：特定日期时间（ISO格式）

   schedule_type = "once"
   schedule_value = "2026-01-06T20:00:00"

GUI任务创建

导航到Gradio GUI中的“任务”选项卡
填写任务详细信息：

- 任务名称 - 选择代理 - 任务提示 - 计划类型和值 - 重复设置

点击“创建任务”

沙盒和安全

沙盒配置

沙盒提供了具有资源限制的执行隔离：

from core.sandbox import SandboxConfig, SandboxManager

config = SandboxConfig(
    enabled=True,
    max_memory_mb=512,
    max_cpu_time=30,
    max_wall_time=60,
    network_access=True,
)

安全策略

from core.security import SecurityPolicy, SecurityValidator

policy = SecurityPolicy(
    allow_command_execution=False,
    allow_file_write=True,
    allow_network_access=True,
    blocked_domains=["malware.com"],
)

validator = SecurityValidator(policy)

# Validate input
valid, error = validator.validate_input(user_input)
if not valid:
    print(f"Blocked: {error}")

受保护的路径

以下路径将自动受到保护：

/etc/passwd, /etc/shadow, /etc/sudoers
~/.ssh, ~/.gnupg
.env, .aws/credentials
包含以下内容的文件： secret, password, token

发展

运行测试

# Unit tests
pytest tests/

# Integration tests
python tests/test_integration.py

# Specific module tests
python tests/test_config.py
python tests/test_agent_manager.py
python tests/test_task_scheduler.py
python tests/test_sandbox.py

代码质量

# Format code
black src/ tests/

# Lint code
ruff check src/ tests/

# Type checking
mypy src/

项目结构

agent-infra/
├── config/                 # Configuration files
│   ├── llm.yaml           # LLM provider settings
│   ├── agents.yaml        # Agent definitions
│   ├── mcp_servers.yaml   # MCP server configs
│   ├── app.yaml           # Application settings
│   └── storage.yaml       # Storage & cache configuration
├── docs/                  # Documentation
│   ├── MCP_SERVER_CONFIGURATION.md  # MCP server configuration guide
│   └── MCP_TROUBLESHOOTING.md  # MCP server troubleshooting guide
├── src/
│   ├── core/              # Core components
│   │   ├── config.py      # Configuration management
│   │   ├── agent_manager.py  # Agent lifecycle (with SDK factory)
│   │   ├── agent_adapter.py  # Unified agent adapter interface
│   │   ├── qwen_agent_adapter.py  # Qwen SDK adapter
│   │   ├── claude_agent_adapter.py # Claude SDK adapter (Computer Use)
│   │   ├── task_scheduler.py  # Task scheduling
│   │   ├── mcp_bridge.py  # MCP integration (stdio + SSE)
│   │   ├── mcp_stdio_client.py  # MCP stdio client
│   │   ├── mcp_sse_client.py    # MCP SSE client (streaming)
│   │   ├── mcp_tool_converter.py  # MCP tool format converter
│   │   ├── sandbox.py     # Sandbox environment
│   │   ├── resource_limiter.py  # Resource limits
│   │   ├── security.py    # Security policies
│   │   └── storage_helpers.py  # Storage adapter helpers
│   ├── storage/           # Storage Adapter Layer
│   │   ├── base_adapter.py      # Storage adapter interface
│   │   ├── base_cache.py        # Cache adapter interface
│   │   ├── base_vector_store.py # Vector store interface (RAG)
│   │   ├── factory.py           # Adapter factory
│   │   ├── config.py            # Storage configuration models
│   │   ├── sqlite_adapter.py     # SQLite implementation
│   │   ├── postgres_adapter.py   # PostgreSQL implementation
│   │   └── redis_cache.py        # Cache implementations
│   ├── agents/            # Agent implementations
│   │   └── base_agent.py  # Base agent class
│   ├── api/               # REST API
│   │   ├── openapi_server.py  # FastAPI server
│   │   ├── sse_endpoints.py    # SSE endpoints for streaming
│   │   └── websocket_endpoints.py  # WebSocket endpoints
│   └── gui/               # Web interface
│       └── app.py         # Gradio GUI
├── storage/               # Local storage (legacy)
│   ├── tasks/             # Task persistence
│   └── logs/              # Application logs
├── tests/                 # Test suite
│   ├── test_*.py          # Unit tests
│   ├── test_storage_adapters.py  # Storage layer tests
│   └── test_integration.py # Integration tests
├── worklogs/              # Development logs
│   ├── storage-adapter-layer/  # Storage adapter implementation
│   │   └── *.md          # Phase reports
│   └── sse-mcp-transport/  # SSE transport implementation
│       └── IMPLEMENTATION.md  # Implementation report
├── plans/                 # Implementation plans
│   └── storage-adapter-layer.md  # Storage adapter plan
├── main.py                # Application entry point
├── requirements.txt       # Python dependencies
├── pyproject.toml        # Project metadata
└── README.md             # This file

故障排除

MCP服务器问题

如果遇到MCP服务器连接问题，请参阅 docs/MCP_TRUBLESHOOTING.md 用于：

连接超时问题
SDK版本兼容性
服务器特定问题
常见错误消息和解决方案

依赖项

核心依赖关系

qwen-agent -Qwen代理SDK
fastapi -REST API框架
gradio -Web UI框架
apscheduler -任务调度
pydantic -数据验证
loguru -日志记录

MCP依赖关系

mcp==1.8.1 -模型上下文协议SDK

- 重要：1.9.0+版本存在已知问题 stdio_client （BrokenResourceError，会话初始化挂起） - 使用v1.8.1进行稳定的stdio传输通信

anyio -异步IO

存储依赖关系

个人版（SQLite）:

aiosqlite -异步SQLite适配器

企业版（PostgreSQL+Redis）:

sqlalchemy -SQL ORM和工具包
asyncpg -异步PostgreSQL驱动程序
redis -用于分布式缓存的Redis客户端

公用事业依赖关系

psutil -资源监控
pyyaml -YAML配置
python-dotenv -环境变量

贡献

分叉存储库
创建要素分支(git checkout -b feature/amazing-feature)
提交您的更改(git commit -m 'Add amazing feature')
推到分支(git push origin feature/amazing-feature)
打开拉取请求

许可证

MIT许可证-有关详细信息，请参阅许可证文件。

致谢

Qwen代理SDK用于代理框架
服务器集成的模型上下文协议
FastAPI和Gradio用于web界面
用于任务调度的APScheduler

Agent Infra Dev

快速接入

详细介绍