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

举报
ceshiren001 发表于 2025/08/22 20:29:52 2025/08/22
【摘要】 本教程带你从零动手,将高德地图和arXiv API快速封装成AI可调用的MCP工具。内容涵盖环境搭建、服务器骨架搭建、工具实现和错误处理,并通过具体代码示例,教你如何用Python创建地理编码和论文检索工具。最后还介绍了使用MCP CLI测试及集成到Claude桌面的详细方法,助你快速构建属于自己的AI工具库。

​本文是 MCP(Model Context Protocol)教程系列的第二阶段,我们将告别理论,直接进入开发实战。你将学会如何把第三方 API快速封装成 Claude 等 AI 模型可直接调用的强大工具。我们将以高德地图地理编码和arXiv 论文检索这两个实用场景为例,内容涵盖资源定义、工具实现、错误处理等核心开发环节,带你一步步完成从零到可用的工具集成。

一、项目初始化与环境搭建

首先,确保你已准备好开发环境。我们推荐使用 Python,因为其生态拥有最完善的 MCP 支持。

  1. 创建项目目录

    mkdir my-mcp-server
    cd my-mcp-server
  2. 创建虚拟环境并激活

    python -m venv venv
    # On Windows
    .\venv\Scripts\activate
    # On macOS/Linux
    source venv/bin/activate
  3. 安装核心依赖: MCP 的核心是 mcp 库,此外我们还需要用于 HTTP 请求的 aiohttp 和用于管理异步并发的 anyio

    pip install mcp aiohttp anyio
  4. 创建主文件

    touch server.py

二、MCP 服务器骨架:起点

每个 MCP Server 都需要一个基本的程序结构。我们在 server.py 中搭建骨架。

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())

现在,你可以运行 python server.py,虽然它还做不了任何事情,但骨架已经搭好。接下来我们为其注入灵魂。

三、实战一:封装高德地图地理编码 API(Tools)

我们将把高德地图的“地理编码”API 封装成一个 MCP Tool,让 Claude 能够根据地址查询经纬度。

1. 获取 API Key

前往高德开放平台,注册账号并创建一个新应用,获取你的 API Key(your_amap_api_key_here)。

2. 声明并实现工具

我们在 app 实例上使用 @app.tool() 装饰器来声明一个工具。

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)

代码解析与最佳实践

  • 工具声明@app.tool() 装饰器将函数注册为 MCP 工具。函数的参数(address: str)和文档字符串("""...""")至关重要,它们会自动成为工具 Schema 的一部分,帮助 Claude 理解如何调用它。
  • 错误处理
    • asyncio.timeout(10):确保网络请求不会无限期挂起。
    • response.raise_for_status():处理非 200 的 HTTP 状态码。
    • 检查高德 API 返回的 status 字段,处理业务逻辑错误。
    • 使用多重 except 块捕获不同类型的异常,并返回清晰的错误信息。
  • 返回值:必须返回 ToolResult 对象。content 是给模型看的结果,isError=True 用于明确告知模型此次调用失败了。

四、实战二:封装 arXiv 论文检索(Resources & Tools)

MCP 不仅有工具(Tools),还有资源(Resources)。资源代表模型可以“读取”的数据源。arXiv 是一个完美的例子,我们可以提供一个资源列表(论文列表),并提供一个工具来搜索它。

1. 定义 arXiv 资源(Resources)

资源使用 @app.list_resources() 和 @app.read_resource() 装饰器。

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
    )

2. 实现 arXiv 搜索工具(Tools)

现在实现核心的搜索功能。

@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)

代码解析

  • 资源与工具的结合:理想情况下,search_arxiv 工具执行搜索后,应该将结果写入 arxiv://search 资源,然后 read_arxiv_resource 函数可以返回格式化后的结果。本例为简化流程,直接在工具调用中返回了结果。这种“混合”模式在实践中也很常见。
  • 参数设计:工具参数 max_results: int = 5 设置了默认值,这让模型在调用时更加灵活。

五、测试与集成

1. 使用 MCP CLI 测试

在开发过程中,你可以使用官方 mcp CLI 工具来测试你的服务器,而无需启动完整的 Claude 环境。

首先全局安装 CLI:

pip install model-context-protocol

然后在你的项目目录下运行:

mcp dev server.py

这会启动一个交互式会话,你可以使用 list_tools, call_tool 等命令来测试你的工具是否正常工作。

2. 集成到 Claude

目前,集成到 Claude for Desktop 是体验 MCP 的最佳方式。

  1. 创建配置文件: 在 ~/.anthropic/(macOS/Linux)或 %USERPROFILE%\.anthropic\(Windows)目录下创建或编辑 claude_desktop_config.json

  2. 配置你的 Server: 将你的服务器路径添加到配置文件中。

    {
      "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"]
        }
      }
    }
  3. 重启 Claude:重启 Claude for Desktop 应用,你现在就可以在对话中直接使用你自定义的工具了!例如:

    “@geocode_address 帮我查一下深圳腾讯大厦的经纬度。”

    “@search_arxiv 帮我找一些关于 diffusion model 的最新论文,找3篇就行。”

总结与进阶

通过本教程,你已经成功完成了:

  1. 项目初始化:搭建了一个标准的 MCP 服务器开发环境。
  2. 工具封装:将高德地图 Geocoding API 封装成了一个可靠的 MCP Tool,并实施了全面的错误处理与超时控制
  3. 资源定义:理解了 Resource 的概念,并为 arXiv 论文库声明了资源模板。
  4. 混合实践:实现了另一个 arXiv 搜索工具,演示了如何设计工具参数。
  5. 测试与集成:学会了如何使用 MCP CLI 测试,并如何将其集成到 Claude 桌面端中。

下一步挑战

  • 完善 arXiv 工具:使用 xml.etree.ElementTree 解析 arXiv 的 XML 响应,并返回格式清晰、包含标题、作者、摘要和链接的论文列表。
  • 状态管理:让你的 read_arxiv_resource 能够返回最近一次搜索的真实结果,实现真正的 Resource 功能。
  • 更多 API:尝试封装其他你常用的 API,如 GitHub、JIRA、Notion、数据库等。
  • 安全性:将 API Keys 等敏感信息移至环境变量中管理。

MCP 的强大在于它能将 AI 模型的能力无缝扩展至整个数字世界。如今,你已具备了创建这些连接器的基本技能,开始构建属于你自己的智能助手工具箱吧!

【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。