MCP 开发实践与示例代码

from fastmcp import mcp, tool
import httpx

@tool(name="get_alerts", description="查询指定地区的天气预警")
async def get_alerts(zone: str):
    async with httpx.AsyncClient() as client:
        url = f"https://api.weather.gov/alerts/active/zone/{zone}"
        resp = await client.get(url)
        data = resp.json()
        return {"type": "text", "text": f"{len(data['features'])} 条预警"}

@tool(name="get_forecast", description="获取七天天气预报")
async def get_forecast(lat: float, lon: float):
    async with httpx.AsyncClient() as client:
        url = f"https://api.weather.gov/points/{lat},{lon}/forecast"
        resp = await client.get(url)
        forecast = resp.json()['properties']['periods']
        text = "\n".join([f"{p['name']}: {p['detailedForecast']}" for p in forecast])
        return {"type": "text", "text": text}

if __name__ == "__main__":
    mcp.run(transport="stdio")

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/allowed/path"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {"GITHUB_PERSONAL_ACCESS_TOKEN": "<你的 token>"}
    }
  }
}

{
  "model": "claude-3-sonnet-20240229",
  "messages": [...],
  "mcp_servers": [
    {
      "url": "https://mcp.asana.com/sse",
      "name": "asana",
      "allowed_tools": ["create_task", "list_tasks"]
    }
  ]
}

from fastmcp.server import MCPServer, tool, resource

@resource(name="project_files", description="用户项目目录", path="/home/user/project", writable=False)
def project_files(): pass

@tool(name="search_in_file", description="在文件中搜索关键词")
def search_in_file(filename: str, keyword: str) -> str:
    with open(filename) as f:
        lines = [line for line in f if keyword in line]
    return "\n".join(lines) or "未找到"

if __name__ == "__main__":
    server = MCPServer(tools=[search_in_file], resources=[project_files])
    server.run_stdio()