MCP AI Infra Real Time Agent

🚀 MCP项目

本项目围绕MCP（模型上下文协议）展开，构建了一套灵活的系统架构。通过MCP协议，实现了AI与外部设备的高效连接，能在一次开发后集成到多个宿主系统。同时，结合容器化技术，简化了项目的创建与更新流程。

🚀 快速开始

MCP协议与通用AI接口

MCP类似于USB - C，MCP服务器如同外部设备，可连接到AI（如Claude Desktop）或云应用程序。借助此特性，我们只需进行一次开发，就能将功能集成到多个宿主系统中。 MCP客户端嵌入在AI宿主（如Claude Desktop、Cursor等）内部，负责与MCP服务器进行1:1交互。MCP客户端具备调用工具、查询资源和处理提示的能力；而MCP服务器则会暴露工具（模型控制：检索、数据库更新、发送）、资源（应用程序控制：数据库记录、API）和提示（用户控制：文档）。

MCP与容器化

• 利用Udemy创建项目，设置虚拟环境，安装依赖项（如MCP CLI），并使用Cursor索引官方的MCP文档。
• 更新项目以符合Cursor规则。

✨ 主要特性

• 通用性强：MCP协议如同USB - C，一次开发可集成到多个宿主系统。
• 交互灵活：MCP客户端与服务器1:1交互，可调用工具、查询资源和处理提示。
• 便于开发：结合容器化技术，通过Udemy创建项目，利用Cursor索引文档，简化开发流程。

📦 安装指南

• 使用Udemy创建项目。
• 设置虚拟环境。
• 安装依赖项，如MCP CLI。

💻 使用示例

基础用法

实现一个简单的MCP服务器 @server.py

# server.py 实现一个简单的MCP服务器，使用MCP Python SDK，并暴露一个名为terminal的工具，允许用户运行终端命令。

from mcp.mcp import MCPClient, MCPServer
import subprocess

class TerminalTool:
    def __init__(self):
        pass
    
    async def run_command(self, command):
        # 使用subprocess执行终端命令并捕获输出
        process = await asyncio.create_subprocess_shell(
            command,
            stdout=asyncio.PIPE,
            stderr=asyncio.PIPE
        )
        stdout, stderr = await process.communicate()
        result = {
            'command': command,
            'stdout': stdout.decode('utf-8'),
            'stderr': stderr.decode('utf-8'),
            'exit_code': process.returncode
        }
        return result

if __name__ == "__main__":
    # 初始化MCP服务器
    server = MCPServer()
    # 注册terminal工具到服务器
    server.register_tool('terminal', TerminalTool())
    
    print("启动MCP服务器...")
    server.start()

高级用法

在MCP服务器中暴露资源

# server.py 暴露位于桌面目录下的mcpreadme.md文件作为资源，使用MCP Python SDK。

from mcp.mcp import MCPClient, MCPServer
import os

class ReadmeResource:
    def __init__(self):
        self.resource_path = os.path.join(os.path.expanduser('~'), 'Desktop', 'mcpreadme.md')
    
    async def get_readme(self):
        try:
            with open(self.resource_path, 'r') as file:
                content = await file.read()
            return {'content': content}
        except FileNotFoundError:
            return {'error': f'文件 {self.resource_path} 未找到'}
        except IOError:
            return {'error': '读取文件时发生错误'}

if __name__ == "__main__":
    # 初始化MCP服务器
    server = MCPServer()
    # 注册readme资源到服务器
    server.register_resource('mcp_readme', ReadmeResource())
    
    print("启动MCP服务器...")
    server.start()

📚 详细文档

开发建议

• 避免重复造轮子：使用现有的工具和库来简化开发流程。
• 调试与测试：利用Anthropic的MCP Inspector进行交互式测试和调试，而无需完整的部署。

参考资料

• MCP速成课程：一日掌握模型上下文协议