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速成課程：一日掌握模型上下文協議