MCP File Operations Agent

MCP ファイル操作エージェント

サーバーとクライアントのアーキテクチャを通じてファイルシステム操作を提供する、シンプルな Model Context Protocol（MCP）の実装です。

コンポーネント

• file_agent_server.py：ファイルシステム操作をツールとして公開する MCP サーバー
• file_agent_client.py：Google の Gemini API を使用するクライアント実装

機能

• ファイルの内容を読み取る
• ファイルに内容を書き込む
• ディレクトリの内容を一覧表示する

要件

• Python ≥3.12
• mcp[cli]>=1.6.0
• google-genai>=1.12.1
• python-dotenv

セットアップ

1. 依存関係をインストールします：pip install -e .
2. API キーを記載した .env ファイルを作成します（.env-sample を参照してください）
3. サーバーを実行します：python file_agent_server.py
4. クライアントを実行します：python file_agent_client.py
5. SSE の例を実行します：python sse_echo.py

MCP エージェントの実装

異なるトランスポートレイヤーを使用して MCP サーバーとクライアントをセットアップする方法を紹介します。

Stdio トランスポート（標準入力／標準出力）

Stdio トランスポートはローカルのプロセス間通信に便利で、クライアントアプリケーションがサーバープロセスを起動する場合によく使用されます。

サーバー（file_agent_server.py の例）：

# file_agent_server.py
from mcp.server.fastmcp import FastMCP
import os # Example tool dependencies

# 1. Create a FastMCP server instance
file_agent = FastMCP("File Operations Agent")

# 2. Define tools using the @file_agent.tool() decorator
@file_agent.tool()
def read_file(file_path: str) -> str:
    """Read the contents of a file."""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        return f"Error reading file: {str(e)}"

# Add other tools (write_file, list_directory) similarly...

# 3. Run the server (automatically uses Stdio)
if __name__ == "__main__":
    file_agent.run()

クライアント（file_agent_client.py の例）：

# file_agent_client.py
import asyncio
from mcp import ClientSession, StdioServerParameters, types as mcp_types
from mcp.client.stdio import stdio_client

# 1. Define server parameters (how to start the server process)
server_params = StdioServerParameters(
    command="python", # Command to execute
    args=["file_agent_server.py"], # Arguments for the command
)

async def run_client():
    try:
        # 2. Use stdio_client to start the server process and get read/write streams
        async with stdio_client(server_params) as (read, write):
            print("stdio_client connected.")
            # 3. Create a ClientSession with the streams
            # Optionally pass a sampling_callback for agent logic
            async with ClientSession(read, write) as session:
                print("ClientSession created.")
                # 4. Initialize the connection
                await session.initialize()
                print("Session initialized.")

                # 5. List available tools
                list_tools_result = await session.list_tools()
                print("Available tools:", list_tools_result.tools)

                # 6. Call a tool
                call_result = await session.call_tool("read_file", {"file_path": "README.md"})
                if call_result.content and isinstance(call_result.content[0], mcp_types.TextContent):
                    print("Read file result:", call_result.content[0].text[:100] + "...")
                else:
                    print("Error or unexpected result:", call_result)

    except Exception as e:
        print(f"Client error: {e}")

if __name__ == "__main__":
    asyncio.run(run_client())

SSE トランスポート（Server-Sent Events）

SSE トランスポートは通信に HTTP を使用し、ウェブベースまたはネットワークベースのやり取りに適しています。

サーバー（sse_echo.py の例）：

# sse_server_example.py (based on sse_echo.py)
import asyncio
from mcp import FastMCP
from mcp.types import TextContent # Example type import

# Define connection details
HOST = "127.0.0.1"
PORT = 8765
SSE_PATH = "/sse" # The HTTP path for the SSE endpoint

# 1. Instantiate FastMCP with SSE configuration
app = FastMCP(name="SSE Echo Server", host=HOST, port=PORT, sse_path=SSE_PATH)

# 2. Define tools as usual
@app.tool()
async def echo(text: str) -> list[TextContent]:
    """Echoes the input text back."""
    print(f"Server received: {text}")
    return [TextContent(type="text", text=f"Echo: {text}")]

# 3. Run the server asynchronously using run_sse_async
async def run_server():
    print(f"Starting SSE server on http://{HOST}:{PORT}{SSE_PATH}")
    await app.run_sse_async()

if __name__ == "__main__":
    asyncio.run(run_server())

クライアント（sse_echo.py の例）：

# sse_client_example.py (based on sse_echo.py)
import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client # SSE specific client helper
from mcp.types import TextContent, CallToolResult # Example type imports

# Server connection details (must match the server)
HOST = "127.0.0.1"
PORT = 8765
SSE_PATH = "/sse"

async def run_client():
    server_url = f"http://{HOST}:{PORT}{SSE_PATH}"
    print(f"Client connecting to {server_url}")

    try:
        # Allow server time to start (in combined demo scripts)
        await asyncio.sleep(1)

        # 1. Use sse_client helper to connect to the server URL
        async with sse_client(url=server_url) as (read, write):
             print("sse_client connected.")
             # 2. Create ClientSession with the read/write streams
             async with ClientSession(read, write) as session:
                print("ClientSession created.")
                # 3. Initialize the session
                await session.initialize()
                print("Session initialized.")

                # 4. Call a tool
                message = "Hello via SSE!"
                response: CallToolResult = await session.call_tool("echo", {"text": message})

                # Process response
                if response.content and isinstance(response.content[0], TextContent):
                     print(f"Client received: {response.content[0].text}")
                else:
                     print(f"Client received unexpected response: {response}")

    except Exception as e:
        print(f"Client error: {e}")

if __name__ == "__main__":
    asyncio.run(run_client())

Cursor IDE との統合（サンプル）

.cursor/mcp.json で設定します：

{
  "mcpServers": {
    "File Operations Agent": {
      "command": "python",
      "args": ["path/to/file_agent_server.py"],
      "env": {}
    }
  }
}