MCP AI Infra Real Time Agent

🚀 プロジェクト概要

このプロジェクトは、MCP（Model Context Protocol）を用いて、AIと外部システムとの連携を容易にする仕組みを提供します。MCPを利用することで、一度開発した機能を複数のAIホストシステムに統合することが可能です。

🚀 クイックスタート

部分1：MCPプロトコルと汎用AIインターフェース

• MCPはUSB - Cに似た存在で、MCPサーバーは外部デバイスのように機能し、AI（Claude Desktopなど）やクラウドアプリケーションに接続できます。一度開発した機能を複数のホストシステムに統合することができます。
• MCPクライアントはAIホスト（Claude Desktop、Cursorなど）内部に埋め込まれ、MCPサーバーと1対1で通信します。MCPクライアントはツールの呼び出し、リソースの照会、プロンプトの処理が可能です。一方、MCPサーバーはツール（モデル制御：検索、データベース更新、送信）、リソース（アプリケーション制御：データベースレコード、API）、プロンプト（ユーザー制御：ドキュメント）を公開します。

部分2：MCPとコンテナ化

• Udemyを使用してプロジェクトを作成し、仮想環境を設定し、依存関係（MCP CLIなど）をインストールし、Cursorを使用して公式のMCPドキュメントをインデックス化します。
• プロジェクトをCursorのルールに合わせて更新します。

✨ 主な機能

• MCPを使用することで、一度開発した機能を複数のAIホストシステムに容易に統合できます。
• MCPサーバーとクライアントの仕組みにより、AIと外部システムの連携を効率的に行えます。

💻 使用例

基本的な使用法

1. 簡単な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()

2. 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速成课程：一日掌握模型上下文协议