MCP 开发实战：手把手教你封装高德地图与 arXiv API

    mkdir my-mcp-server    cd my-mcp-server

    python -m venv venv    # On Windows    .\venv\Scripts\activate    # On macOS/Linux    source venv/bin/activate

    pip install mcp aiohttp anyio

    touch server.py

import asynciofrom mcp.server import Serverfrom mcp.server.stdio import stdio_server
# 创建 Server 实例，名字叫 "my-custom-tools"app = Server("my-custom-tools")
# 此处将在这里注册我们的工具（Tools）和资源（Resources）
async def main():    # 使用 stdio 传输层，这是与 Claude 等客户端通信的标准方式    async with stdio_server() as (read_stream, write_stream):        await app.run(            read_stream,            write_stream,            app.create_initialization_options()        )
if __name__ == "__main__":    asyncio.run(main())

import aiohttpfrom mcp.server.models import ToolResultfrom mcp.types import Tool
# 你的高德 API Key，在实际项目中应从环境变量读取！AMAP_API_KEY = "your_amap_api_key_here" 
@app.tool()async def geocode_address(address: str) -> Tool:    """根据中文地址获取其经纬度坐标。
    Args:        address: 详细的中文地址，例如：北京市朝阳区阜通东大街6号。
    Returns:        返回该地址的经纬度信息。    """    # 构建请求 URL    url = f"https://restapi.amap.com/v3/geocode/geo?key={AMAP_API_KEY}&address={address}"
    # 错误处理与超时控制：使用 aiohttp 和 asyncio.timeout    try:        async with aiohttp.ClientSession() as session:            # 设置 10 秒超时            async with asyncio.timeout(10):                async with session.get(url) as response:                    # 检查 HTTP 状态码是否成功                    response.raise_for_status()                    data = await response.json()
                    # 检查高德 API 返回的业务状态码                    if data.get('status') != '1':                        error_info = data.get('info', 'Unknown error')                        return ToolResult(                            content=f"Geocoding API error: {error_info}",                            isError=True                        )
                    # 解析并返回结果                    geocodes = data.get('geocodes', [])                    if not geocodes:                        return ToolResult(content="No results found for the given address.")
                    location = geocodes[0].get('location')                    formatted_address = geocodes[0].get('formatted_address')                    result_text = f"地址 '{formatted_address}' 的坐标是：{location}"
                    return ToolResult(content=result_text)
    except asyncio.TimeoutError:        return ToolResult(content="Geocoding request timed out after 10 seconds.", isError=True)    except aiohttp.ClientResponseError as e:        return ToolResult(content=f"HTTP error occurred: {e.status} - {e.message}", isError=True)    except Exception as e:        return ToolResult(content=f"An unexpected error occurred: {str(e)}", isError=True)

from mcp.server.models import ResourceTemplatesResult, ReadResourceResultfrom mcp.types import ResourceTemplate
@app.list_resources()async def list_arxiv_resources() -> ResourceTemplatesResult:    """列出可用的 arXiv 相关资源。"""    templates = [        ResourceTemplate(            uri="arxiv://search",            name="arXiv Search Results",            description="搜索结果来自 arXiv 论文库",            mimeType="text/plain"        )    ]    return ResourceTemplatesResult(templates=templates)
@app.read_resource()async def read_arxiv_resource(uri: str) -> ReadResourceResult:    """读取 arxiv://search 资源。    注意：这个示例中，我们简单返回一个提示。    在实际应用中，你可能会在这里缓存或返回最近一次搜索的结果。    """    if uri == "arxiv://search":        return ReadResourceResult(            content="Use the 'search_arxiv' tool to perform a search first.",            mimeType="text/plain"        )    # 对于未知的 URI，返回错误    return ReadResourceResult(        content=f"Resource not found: {uri}",        mimeType="text/plain",        isError=True    )

@app.tool()async def search_arxiv(query: str, max_results: int = 5) -> Tool:    """在 arXiv 库中搜索科学论文。
    Args:        query: 搜索关键词，例如：'large language model'。        max_results: 返回的最大结果数量，默认为 5。    """    # 构建 arXiv API 请求 URL    base_url = "http://export.arxiv.org/api/query"    params = {        "search_query": f"all:{query}",        "start": 0,        "max_results": max_results,        "sortBy": "submittedDate",        "sortOrder": "descending"    }
    try:        async with aiohttp.ClientSession() as session:            async with asyncio.timeout(15): # arXiv 有时较慢，设置稍长的超时                async with session.get(base_url, params=params) as response:                    response.raise_for_status()                    data = await response.text()
        # arXiv 返回 Atom XML，这里需要解析（示例中简化处理）        # 在实际项目中，你应该使用 xml.etree.ElementTree 来解析响应内容        # 这里我们简单地截取一部分文本作为演示        if len(data) > 500:            preview = data[:500] + "..."        else:            preview = data
        result_content = f"**arXiv Search Results for '{query}':**\n\nRaw API response preview:\n{preview}\n\n*Note: This is a simplified demo. A real implementation would parse the XML and present a formatted list of papers.*"
        return ToolResult(content=result_content)
    except asyncio.TimeoutError:        return ToolResult(content="arXiv search request timed out after 15 seconds.", isError=True)    except aiohttp.ClientResponseError as e:        return ToolResult(content=f"HTTP error occurred: {e.status} - {e.message}", isError=True)    except Exception as e:        return ToolResult(content=f"An unexpected error occurred during arXiv search: {str(e)}", isError=True)

pip install model-context-protocol

mcp dev server.py

    {      "mcpServers": {        "my-custom-tools": {          "command": "/path/to/your/venv/bin/python",          "args": ["/absolute/path/to/your/project/server.py"],          "env": {"AMAP_API_KEY": "your_actual_key_here"} // 推荐从环境变量读取密钥！        }      }    }

*Windows 示例*：

    {      "mcpServers": {        "my-custom-tools": {          "command": "C:\\path\\to\\your\\project\\venv\\Scripts\\python.exe",          "args": ["C:\\path\\to\\your\\project\\server.py"]        }      }    }