MCP File Operations Agent

MCP 文件操作 Agent

一个简单的模型上下文协议（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 Agent

以下介绍如何使用不同的传输层设置 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 传输（服务器发送事件）

SSE 传输使用 HTTP 进行通信，适用于基于 Web 或网络的交互。

服务端（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": {}
    }
  }
}