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

简介: 本教程为 MCP(Model Context Protocol)开发实战第二阶段,带你从零封装第三方 API 为 AI 模型可用工具。通过高德地图地理编码与 arXiv 论文检索两个实例,涵盖项目搭建、工具声明、资源定义、错误处理等核心内容,助你快速上手 MCP 开发并集成至 Claude 使用。

本文是 MCP(Model Context Protocol)教程系列的第二阶段,我们将告别理论,直接进入开发实战。你将学习如何从零开始,将一个第三方 API 封装成 Claude 等 AI 模型可以直接调用的强大工具。我们将以高德地图地理编码和arXiv 论文检索这两个实用场景为例,完整覆盖资源定义、工具声明、错误处理等核心概念。

一、项目初始化与环境搭建
首先,确保你已准备好开发环境。我们推荐使用 Python,因为其生态拥有最完善的 MCP 支持。

创建项目目录:

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

python -m venv venv

On Windows

.\venv\Scripts\activate

On macOS/Linux

source venv/bin/activate
安装核心依赖: MCP 的核心是 mcp 库,此外我们还需要用于 HTTP 请求的 aiohttp 和用于管理异步并发的 anyio。

pip install mcp aiohttp anyio
创建主文件:

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
)
  1. 实现 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 等命令来测试你的工具是否正常工作。

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

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

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

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

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

总结与进阶
通过本教程,你已经成功完成了:

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

完善 arXiv 工具:使用 xml.etree.ElementTree 解析 arXiv 的 XML 响应,并返回格式清晰、包含标题、作者、摘要和链接的论文列表。
状态管理:让你的 read_arxiv_resource 能够返回最近一次搜索的真实结果,实现真正的 Resource 功能。
更多 API:尝试封装其他你常用的 API,如 GitHub、JIRA、Notion、数据库等。
安全性:将 API Keys 等敏感信息移至环境变量中管理。
MCP 的强大之处在于它将 AI 模型的能力无限延伸至整个数字世界。现在,你已经掌握了打造这些连接器的基本技能,开始构建你自己的智能助手工具箱吧!

相关文章
|
11天前
|
供应链 监控 安全
1688商品详情API接口实战指南:合规获取数据,驱动B2B业务增长
1688商品详情API(alibaba.product.get)是合规获取B2B商品数据的核心工具,支持全维度信息调用,助力企业实现智能选品、供应链优化与市场洞察,推动数字化转型。
|
16天前
|
数据可视化 测试技术 API
从接口性能到稳定性:这些API调试工具,让你的开发过程事半功倍
在软件开发中,接口调试与测试对接口性能、稳定性、准确性及团队协作至关重要。随着开发节奏加快,传统方式已难满足需求,专业API工具成为首选。本文介绍了Apifox、Postman、YApi、SoapUI、JMeter、Swagger等主流工具,对比其功能与适用场景,并推荐Apifox作为集成度高、支持中文、可视化强的一体化解决方案,助力提升API开发与测试效率。
|
7天前
|
JSON API 网络架构
API协议全景图:从REST到MCP的选型指南
本文以开源项目 HiMarket 为背景,系统梳理了从传统Web应用到现代AI场景下的六种主流API协议,帮助开发者厘清不同协议的核心定位、技术特点与适用场景。
|
16天前
|
JSON 数据挖掘 API
微店商品详情API接口开发指南:从零到实战
微店商品详情API(micro.item_get)用于获取商品名称、价格、库存等信息,支持HTTP GET/POST请求,返回JSON格式数据,适用于电商开发、店铺管理与数据分析。提供Python请求示例,便于快速集成调用,适用于多店铺管理、跨平台展示及价格监控等场景。
|
7天前
|
数据采集 缓存 API
小红书笔记详情 API 实战指南:从开发对接、场景落地到收益挖掘(附避坑技巧)
本文详解小红书笔记详情API的开发对接、实战场景与收益模式,涵盖注册避坑、签名生成、数据解析全流程,并分享品牌营销、内容创作、SAAS工具等落地应用,助力开发者高效掘金“种草经济”。
小红书笔记详情 API 实战指南:从开发对接、场景落地到收益挖掘(附避坑技巧)
|
11天前
|
缓存 监控 供应链
亚马逊 MWS API 实战:商品详情精准获取与跨境电商数据整合方案
本文详细解析亚马逊MWS API接口的技术实现,重点解决跨境商品数据获取中的核心问题。文章首先介绍MWS接口体系的特点,包括多站点数据获取、AWS签名认证等关键环节,并对比普通电商接口的差异。随后深入拆解API调用全流程,提供签名工具类、多站点客户端等可复用代码。针对跨境业务场景,文章还给出数据整合工具实现方案,支持缓存、批量处理等功能。最后通过实战示例展示多站点商品对比和批量选品分析的应用,并附常见问题解决方案。该技术方案可直接应用于跨境选品、价格监控等业务场景,帮助开发者高效获取亚马逊商品数据。
|
16天前
|
存储 监控 前端开发
淘宝商品详情 API 实战:5 大策略提升店铺转化率(附签名优化代码 + 避坑指南)
本文深入解析淘宝商品详情API的核心字段与实战应用,分享如何通过动态定价、库存预警、差评控制等5大策略提升电商转化率。结合300+店铺实战经验,提供优化代码与避坑指南,助力开发者与运营者实现数据驱动的精细化运营。
|
4天前
|
JSON API 调度
Midjourney 技术拆解与阿里云开发者实战指南:从扩散模型到 API 批量生成
Midjourney深度解析:基于优化Stable Diffusion,实现文本到图像高效生成。涵盖技术架构、扩散模型原理、API调用、批量生成系统及阿里云生态协同,助力开发者快速落地AIGC图像创作。
|
16天前
|
数据采集 JSON API
微店商品列表API接口开发指南:从零到实战
微店商品列表API(vdian.shop.item.list.get)用于获取店铺商品数据,支持分页、签名认证,返回JSON格式。适用于商品同步、竞品分析、多平台展示及数据清洗。提供Python请求示例,便于快速接入。
|
JSON JavaScript 前端开发
开发API完成,写个文档
Jira对接Prism开发API指南   部门 证系统运维团队 文档制作人 陈刚() 时间 2017-04-05 版本 第一版     目录 目的... 1 通例:... 1 认证... 2 新建版本单... 2 获取指定版本单的发布单信息... 3     目的 为了提升工作效率,打通jira和prism之间的联系,让软件项目管理人员可以在jira上新建版本单,并跟踪发布进度,特在prism上制作相关API供jira调用。
1451 0