MCP 实战指南：高德地图和 arXiv API 封装指导教程

import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server

# 创建 Server 实例，名字叫 "my-custom-tools"
app = Server("my-custom-tools")

# 此处将在这里注册我们的工具（Tools）和资源（Resources）

asyncdef main():
    # 使用 stdio 传输层，这是与 Claude 等客户端通信的标准方式
    asyncwith 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 aiohttp
from mcp.server.models import ToolResult
from mcp.types import Tool

# 你的高德 API Key，在实际项目中应从环境变量读取！
AMAP_API_KEY = "your_amap_api_key_here"

@app.tool()
asyncdef 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:
        asyncwith aiohttp.ClientSession() as session:
            # 设置 10 秒超时
            asyncwith asyncio.timeout(10):
                asyncwith 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', [])
                    ifnot 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, ReadResourceResult
from mcp.types import ResourceTemplate

@app.list_resources()
asyncdef 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()
asyncdef 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()
asyncdef 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:
        asyncwith aiohttp.ClientSession() as session:
            asyncwith asyncio.timeout(15): # arXiv 有时较慢，设置稍长的超时
                asyncwith 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