AI导哈

智能代理系统MCP服务器

一个基于模型上下文协议(MCP)的生产级智能代理系统，提供工具调用、资源访问和提示模板功能，适用于LLM应用开发和任务自动化场景。

  • by joohnnie
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 joohnnie/mcp-agent

使用说明

带智能代理系统的MCP演示服务器

一个生产就绪的演示 模型上下文协议（MCP） 用Python实现的服务器，具有智能 代理系统 具有子代理和编排功能。该项目通过综合示例展示了构建MCP服务器和多代理系统的最佳实践。

什么是MCP？

这 模型上下文协议（MCP） 是一个开放协议，规范了应用程序如何向大型语言模型（LLM）提供上下文。MCP服务器公开：

  • 工具：LLM可以调用的可执行函数（例如计算器、文件操作）
  • 资源LLM可以读取的数据源（例如配置、文档）
  • 提示词：用于常见任务的可重用提示模板

特性

代理系统（新！）

与MCP配合使用的生产就绪智能代理：

  1. 代理 -集成MCP的核心代理

    • 使用MCP工具执行任务
    • 将工作委托给专业分代理
    • 跟踪任务执行和结果
    • 处理错误、重试和超时

  2. 子代理 -特定任务的专业代理

    • CalculatorSubAgent-数学计算
    • FileOperationsSubAgent-文件系统操作
    • WeatherSubAgent-天气信息
    • Timestamp子代理-时间戳操作
    • DataProcessingSubAgent-数据处理管道

  3. 代理编排器 -多智能体协调

    • 管理代理池
    • 自动将任务分配给有能力的代理
    • 执行具有依赖关系的工作流
    • 并行和顺序执行
    • 结果汇总和统计

  4. 任务管理 -强大的任务系统

    • 具有优先级的任务定义
    • 状态跟踪（待定、进行中、已完成、失败）
    • 重试逻辑和超时处理
    • 详细的结果跟踪

MCP服务器工具

  1. 计算器 -基本数学运算

    • 运算：加、减、乘、除
    • 完整的输入验证和错误处理

  2. 文件操作 -文件系统交互

    • 读取、写入、列出目录、检查文件是否存在
    • 带有正确错误消息的安全路径处理

  3. 天气 -模拟天气信息

    • 获取任何城市的天气数据
    • 支持摄氏和华氏温度

  4. 时间戳 -以各种格式获取当前时间

    • ISO格式、Unix时间戳、人类可读格式

可用资源

  1. 服务器配置 (config://server/settings)

    • 当前服务器设置和元数据
    • JSON格式的配置

  2. 系统信息 (system://info)

    • 操作系统、Python版本、工作目录
    • 服务器进程信息

  3. 文档 (docs://getting-started)

    • 入门指南
    • 使用说明

提供提示

  1. 代码审查 -生成代码审查清单

    • 可通过编程语言定制
    • 可调复杂度级别

  2. 文档 -文档模板

    • API、用户和开发人员文档类型
    • 项目特定定制

  3. 调试助手 -调试指导

    • 结构化调试方法
    • 常见技术和最佳实践

安装

先决条件

  • Python 3.10或更高版本
  • pip（Python包管理器）

再进行

# Clone the repository
git clone https://github.com/yourusername/mcp-agent.git
cd mcp-agent

# Install the package
pip install -e .

# Or install dependencies directly
pip install -r requirements.txt

开发设置

# Install development dependencies
pip install -r requirements-dev.txt

# Run tests
pytest

# Format code
black src/

# Lint code
ruff check src/

用法

运行代理系统演示

体验代理系统的全部功能：

# Run comprehensive agent system demo
python examples/agent_demo.py

此演示展示了：

  • 具有基本任务的简单代理
  • 代理与子代理（任务委托）
  • 管理多个代理的编排器
  • 多步骤工作流执行
  • 数据处理流水线

快速代理系统示例

import asyncio
from mcp_demo import Agent, AgentCapability, Task
from mcp_demo import CalculatorSubAgent, FileOperationsSubAgent
from mcp_demo import AgentOrchestrator, Workflow, WorkflowStep

async def main():
    # Create main agent with subagents
    main_agent = Agent(name="MainAgent")
    calc_sub = CalculatorSubAgent(parent_agent=main_agent)
    file_sub = FileOperationsSubAgent(parent_agent=main_agent)

    # Create tasks
    calc_task = Task(
        name="Calculate",
        task_type="calculation",
        parameters={"operation": "add", "a": 10, "b": 20},
    )

    # Execute - automatically delegated to appropriate subagent
    async with main_agent, calc_sub, file_sub:
        result = await main_agent.execute_task(calc_task)
        print(f"Result: {result.data}")

asyncio.run(main())

使用编排器

from mcp_demo import AgentOrchestrator, CalculatorSubAgent, Task

async def orchestrate():
    # Create orchestrator
    orch = AgentOrchestrator(name="MainOrch")

    # Register agents
    orch.register_agent(CalculatorSubAgent())

    # Execute tasks in parallel
    async with orch:
        results = await orch.execute_tasks(
            [task1, task2, task3],
            parallel=True,
        )

        # Get statistics
        stats = orch.get_statistics()
        print(f"Success rate: {stats['overall_success_rate']:.1f}%")

运行MCP服务器

服务器使用stdio进行通信，这是MCP服务器的标准传输方式：

# Run directly with Python
python -m mcp_demo.server

# Or use the installed script
mcp-demo

Claude桌面配置

要将此MCP服务器与Claude Desktop一起使用，请将其添加到您的Claude配置文件中：

macOS: ~/Library/Application Support/Claude/claude_desktop_config.json

视窗: %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "demo": {
      "command": "python",
      "args": [
        "-m",
        "mcp_demo.server"
      ],
      "env": {}
    }
  }
}

或者使用已安装的命令：

{
  "mcpServers": {
    "demo": {
      "command": "mcp-demo",
      "args": [],
      "env": {}
    }
  }
}

与MCP客户端一起使用

您还可以使用附带的示例客户端：

python examples/client.py

项目结构

mcp-agent/
├── src/
│   └── mcp_demo/
│       ├── __init__.py          # Package initialization
│       ├── server.py            # MCP server implementation
│       ├── agent.py             # Core agent with MCP integration
│       ├── subagent.py          # Specialized subagents
│       ├── orchestrator.py      # Multi-agent orchestration
│       └── task.py              # Task management system
├── examples/
│   ├── client.py                # Example MCP client
│   ├── agent_demo.py            # Comprehensive agent system demo
│   └── claude_config.json       # Example Claude Desktop config
├── tests/
│   ├── __init__.py
│   ├── test_server.py           # Server tests
│   └── test_agents.py           # Agent system tests
├── docs/
│   ├── ARCHITECTURE.md          # MCP server architecture
│   └── AGENT_SYSTEM.md          # Agent system documentation
├── pyproject.toml               # Project configuration
├── requirements.txt             # Production dependencies
├── requirements-dev.txt         # Development dependencies
├── README.md                    # This file
└── LICENSE                      # MIT License

代码架构

服务器实现

服务器遵循干净的模块化架构：

class MCPDemoServer:
    """Main server class with handler methods"""

    def __init__(self, name: str):
        """Initialize server and register handlers"""

    async def list_tools(self) -> list[Tool]:
        """Return available tools"""

    async def call_tool(self, name: str, arguments: Any) -> list[TextContent]:
        """Execute a tool"""

    async def list_resources(self) -> list[Resource]:
        """Return available resources"""

    async def read_resource(self, uri: AnyUrl) -> str:
        """Read a resource"""

    async def list_prompts(self) -> list[Prompt]:
        """Return available prompts"""

    async def get_prompt(self, name: str, arguments: dict) -> GetPromptResult:
        """Get a prompt with arguments"""

关键设计模式

  1. 类型安全：Pydantic模型的完整类型提示
  2. 错误处理：使用日志记录进行全面的尝试捕获
  3. 验证：使用Pydantic模式进行输入验证
  4. 日志记录：结构化日志记录到文件和stderr
  5. 异步/等待：始终采用适当的异步模式
  6. 关注点分离：每个工具都有自己的方法

例子

示例1：使用计算器工具

# Tool call from LLM client
{
  "name": "calculator",
  "arguments": {
    "operation": "add",
    "a": 15,
    "b": 27
  }
}

# Response
{
  "content": [
    {
      "type": "text",
      "text": "Result: 15 add 27 = 42"
    }
  ]
}

示例2：阅读资源

# Read server configuration
{
  "uri": "config://server/settings"
}

# Response (JSON)
{
  "server_name": "mcp-demo-server",
  "version": "1.0.0",
  "max_connections": 100,
  "features": ["tools", "resources", "prompts"]
}

示例3：获得提示

# Get code review prompt
{
  "name": "code-review",
  "arguments": {
    "language": "Python",
    "complexity": "complex"
  }
}

# Response: Detailed code review checklist for Python

发展

添加新工具

  1. 使用Pydantic定义输入模式：
class MyToolInput(BaseModel):
    param1: str = Field(description="Description")
    param2: int = Field(default=0, description="Description")
  1. 将工具添加到 list_tools():
Tool(
    name="my_tool",
    description="What this tool does",
    inputSchema=MyToolInput.model_json_schema(),
)
  1. 执行工具逻辑：
async def _my_tool(self, arguments: dict) -> list[TextContent]:
    tool_input = MyToolInput(**arguments)
    # Your implementation here
    return [TextContent(type="text", text="Result")]
  1. 增添 call_tool() 调度员：
if name == "my_tool":
    return await self._my_tool(arguments)

添加新资源

  1. 增添 list_resources():
Resource(
    uri=AnyUrl("my://resource"),
    name="My Resource",
    description="What this resource provides",
    mimeType="application/json",
)
  1. 在中添加处理程序 read_resource():
if uri_str == "my://resource":
    data = {"key": "value"}
    return json.dumps(data, indent=2)

添加新提示

  1. 增添 list_prompts():
Prompt(
    name="my-prompt",
    description="What this prompt does",
    arguments=[
        {
            "name": "arg1",
            "description": "Argument description",
            "required": True,
        }
    ],
)
  1. 在中添加处理程序 get_prompt():
if name == "my-prompt":
    arg1 = arguments.get("arg1")
    message = f"Prompt template with {arg1}"
    return GetPromptResult(
        messages=[
            PromptMessage(
                role="user",
                content=TextContent(type="text", text=message),
            )
        ]
    )

测试

运行综合测试套件：

# Run all tests
pytest

# Run with coverage
pytest --cov=mcp_demo --cov-report=html

# Run specific test files
pytest tests/test_server.py      # MCP server tests
pytest tests/test_agents.py      # Agent system tests

# Run with verbose output
pytest -v

# Run agent tests only
pytest tests/test_agents.py -v

最佳实践演示

  1. 输入验证：所有工具输入均已Pydantic验证
  2. 错误处理：通过有意义的消息进行全面的错误处理
  3. 日志记录：用于调试和监控的结构化日志记录
  4. 类型安全：整个代码库中的完整类型提示
  5. 文档：全面的文档和评论
  6. 测试：所有主要功能的单元测试
  7. 代码质量：黑色，荷叶边
  8. 异步模式：正确使用async/await
  9. 资源管理：适当的清理和资源处理
  10. 安全：具有路径验证的安全文件操作

故障排除

常见问题

问题：找不到模块错误

# Solution: Install in development mode
pip install -e .

问题：服务器未出现在Claude Desktop中

# Solution: Check configuration file path and JSON syntax
# Restart Claude Desktop after configuration changes

问题：导入错误 mcp 包裹

# Solution: Install latest MCP SDK
pip install --upgrade mcp

调试模式

启用调试日志记录：

import logging
logging.basicConfig(level=logging.DEBUG)

检查日志文件：

tail -f mcp_server.log

贡献

欢迎投稿！拜托：

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

文档

资源

许可证

此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。

致谢

  • 与 MCP Python SDK
  • 受MCP社区示例的启发
  • 感谢Anthropic开发了模型上下文协议

支持

MCP服务器建设快乐！ 🚀

相关 MCP