本文详细介绍了如何将MCP（Model Context Protocol）集成到官方Claude Desktop应用中。首先，通过配置`claude_desktop_config.json`文件，将MCP服务器信息告知客户端，并演示了如何使用文件系统作为MCP服务器的接入示例。随后，深入解析了MCP的组成部分，包括Host、Client和Server，以及Resources、Prompts和Tools等核心概念。此外，还探讨了Transports通信机制和Sampling进阶使用。最后，文章提供了MCP Server和Client的简单代码实现，并通过天气查询的例子，演示了如何进行测试和接入Claude Desktop。无论是开发者还是对LLM应用集成感兴趣的用户，都能从中获得实用的指导。

💡MCP由三部分组成：Host（LLM应用，负责服务发现和连接管理）、Client（Host应用内部的中间件，与MCP Server保持连接）和Server（提供提示词、工具和资源，实现具体业务逻辑）。

🧰 MCP Server 提供 Resources（文本和二进制两种资源类型，通过URI进行标注）、Prompts（可复用的提示模板和工作流程）和 Tools（将本地代码功能封装为工具，供客户端调用）三种核心能力。

📡 MCP 主要通过 stdio 标准输入输出和 SSE 两种方式进行通信，消息格式采用 JSON-RPC 形式，包括 Request、Response 和 Notification 三种类型。

🌡️ 通过实现一个简单的天气查询MCP Server，演示了如何利用高德API获取城市天气和代码，并通过`@mcp.tool`注解将方法暴露给客户端，方便LLM调用。

文章主要内容：

介绍官方Claude Desktop使用MCP介绍、一些概念MCP Server的简单实现、测试及代码解释MCP Client的简单实现及代码解释

Claude Desktop示范

首先我们需要一个Claude Desktop应用、和一个可用的authropic账号在你的电脑上，接下来我们以官方的fileSystem为示例言是一个mcp server的接入。

打开路径格式为/Users/zhangmingyuan/Library/Application Support/Claude/claude_desktop_config.json的配置文件

添加如下内容进文件并报存。其中filesystem是服务的名称，command和args分别是指令和参数其中后两个参数。

{  "mcpServers": {    "filesystem": {      "command": "npx",      "args": [        "-y",        "@modelcontextprotocol/server-filesystem",        "/Users/zhangmingyuan/Desktop",        "/Users/zhangmingyuan/Downloads"      ]    }  }}

在将MCP server的信息以配置文件告知客户端后，需要重启客户端使用

此时我们可以在客户端对话框中看到filesystem，这说明mcp server已经可用且对应的客户端已经建立

我们可以直接使用，LLM会按需调用我们提供的工具

MCP

组成

flowchart LR    subgraph "Host"        client1[MCP Client]        client2[MCP Client]    end    subgraph "Server Process"        server1[MCP Server]    end    subgraph "Server Process"        server2[MCP Server]    end    client1 <-->|Transport Layer| server1    client2 <-->|Transport Layer| server2

一个完整的MCP主要是由三部分组成：

host：是发起连接的LLM应用程序；具体作用是服务发现、连接管理；表现形式一般是下载用户侧的客户端，如VsCode、Claude Desktop、cherry studio等client：在host应用的内部，与MCP server通过Transport保持一比一的连接，可以看做LLM应用和MCP server的中间件。server：实现一些提示词、工具、资源(数据或内容)提供给客户端，以供客户端实现调用逻辑。主流server实现的一般是提供一些具体的业务/逻辑功能的实现，作为tool给到客户端。

以上是一个MCP的典型组成，接下来介绍其他核心概念：

mcp server可提供的能力：
ResourcesPromptsTools
Transports：通信机制Root：提供给客户端，使之告知服务器相关资源及位置，定义服务器可以运行的边界。(比较简单，没什么好说的)Sampling：server通过客户端请求LLM进行信息补充

主要功能

Resources

种类及标注格式

Resources允许服务器公开可由客户端读取并用作 LLM 交互上下文的数据和内容，资源种类限于两种

文本二进制

标注格式为：[protocol]://[host]/[path]，示例如下：

file:///home/user/documents/report.pdfpostgres://database/customers/schemascreen://localhost/display1

使用流程

具体使用中的动作包括：资源发现、资源读取和更新资源

资源发现：对应静态资源、动态资源有俩种方式进行资源发现
直接资源：服务器通过 resources/list 端点暴露具体资源列表对于动态资源，服务器可以公开 URI 模板 ，客户端可以使用它来构建有效的资源 URI
读取资源：此过程，客户端需要使用资源 URI 发出 resources/read 请求，服务器以资源内容列表进行响应。更新资源：也是两种机制，一种是全量推送更新、一种是订阅
当服务器可用资源列表发生变化时，服务器通过 notifications/resources/list_changed 全量推送通知客户端。客户端发送 resources/subscribe 资源 URI记行订阅；资源发生变化时，服务器发送notifications/resources/updated；客户端可以通过 resources/read 获取最新内容

Prompts

定义可重复使用的提示模板和工作流程，客户端可以轻松地将其呈现给用户和 LLM。提供一种强大的方法来标准化和共享常见的 LLM 交互。

发现和更新的流程类似Resource。主要是功能的复用

Tools

主流Server中核心能力，将本地代码实现的功能方法/已有业务系统中的功能封装为tools提供给客户端，使得host应用中的LLM可以依据客户端定义的执行逻辑来操作tools

注意事项：需要具备tool calling的LLM才能使用

通信协议Transports

主要通信方式有两种

stdio标准输入输出SSE：使用受限，具体场景：
仅需要服务器到客户端的流式传输使用受限制的网络实现简单更新

消息格式采用JSON-RPC的形式，有三种：

Request

{  jsonrpc: "2.0",  id: number | string,  method: string,  params?: object}

Response

{  jsonrpc: "2.0",  id: number | string,  result?: object,  error?: {    code: number,    message: string,    data?: unknown  }}

Notification

{  jsonrpc: "2.0",  method: string,  params?: object}

其消息连接的生命周期如下图：

sequenceDiagram    participant Client    participant Server    Client->>Server: initialize request    Server->>Client: initialize response    Client->>Server: initialized notification    Note over Client,Server: Connection ready for use

进阶使用Sampling

允许server通过客户端请求LLM进行信息补充，工作流程如下：

服务器向客户端发送 sampling/createMessage 请求客户端审核请求并修改客户端从LLM中采样客户端审核LLM生成的内容客户端返回结果给服务器

并非所有host应用都可以处理采样，对上面其他能力的支持也不尽相同，主流host应用对MCP各功能的支持可以查看这里

MCP Server

下面我们对Server进行简单的代码实现，功能是与官网类似的天气查询，只演示tool的实现

简单实现

使用uv初始化工程后，新建weather.py，引入依赖：

from typing import Anyimport httpxfrom mcp.server.fastmcp import FastMCPimport asyncioimport xml.etree.ElementTree as ET

首先我们进行mcp服务器的初始化：

mcp = FastMCP("china_weather")

然后在后续代码中利用高德的API实现城市天气的查询以及城市代码的获取(自己实现为任何你想实现的功能方法)

为了MCP可以识别，在方法上方添加@mcp.tool的注解；

为了方便后续模型理解，在方法内部添加方法及方法参数的说明；

完善后代码如下(完整代码附在最后)：

@mcp.tool()async def get_city_weather(adcode: str) -> str:    """Get today and 3day weather forecast for a adcode.    Args:        adcode: adcode or city name (e.g. "101010100" for Beijing, or "北京")    """    ... ...    @mcp.tool()async def search_city_code(city_name: str) -> str:    """Search for location information by city name using Amap API.    Args:        city_name: Chinese city name or address (e.g. "北京市朝阳区阜通东大街6号")    """    ... ..

最后在主入口将mcp server启动起来：

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

测试

至此我们的完整工具方法已经实现，工具的正确性我们要先验证下：

npx @modelcontextprotocol/inspector <command>

我们使用官方的工程进行测试，command就是具体的启动命令，若不填入也可在后启动后从页面添加，启动后从控制台找到网址打开：

页面视图如下：

可以通过更改左侧的配置更换其他MCP server，可以通过list tools之后选择tool进行测试。

接入Claude Desktop

测试完毕后，我们可以比照前面的filesystem一样将代码发布到pypl，pyproject.toml的内容：

[project]name = "zimy-demo-weather"version = "0.1.2"description = "输入地址查天气"readme = "README.md"requires-python = ">=3.10"dependencies = [    "httpx>=0.28.1",    "mcp>=1.6.0",][project.scripts]zimy-demo-weather = "main:main"

发布后可用相似的方法添加mcp server到claude_desktop_config.json

或者使用本地接入的方式：

    "weather": {      "command": "/Users/zhangmingyuan/opt/anaconda3/bin/uv",      "args": [        "--directory",        "/Users/zhangmingyuan/WorkSpace/WorkSpace_AI/MCP_DEMO/weather",        "run",        "weather.py"      ],      "env": {        "PYTHONPATH": "/Users/zhangmingyuan/WorkSpace/WorkSpace_AI/MCP_DEMO/weather/"      }    },

试一下：

MCP client

作为一个通用host应用，claude提供的client的处理一般是直接将可用的tools提供给大模型，由其自主调用和安排执行逻辑，如果我们想自己设置一些固定的处理，或者对server提供的tools进行一些进一步的处理，就需要自主开发客户端甚至是host

简单实现

这次我们直接上官方的示例代码

import asynciofrom typing import Optionalfrom contextlib import AsyncExitStackfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom anthropic import Anthropicfrom dotenv import load_dotenvload_dotenv()  # load environment variables from .envclass MCPClient:    def __init__(self):        # Initialize session and client objects        self.session: Optional[ClientSession] = None        self.exit_stack = AsyncExitStack()        self.anthropic = Anthropic()    async def connect_to_server(self, server_script_path: str):        """Connect to an MCP server                Args:            server_script_path: Path to the server script (.py or .js)        """        is_python = server_script_path.endswith('.py')        is_js = server_script_path.endswith('.js')        if not (is_python or is_js):            raise ValueError("Server script must be a .py or .js file")                    command = "python" if is_python else "node"        server_params = StdioServerParameters(            command=command,            args=[server_script_path],            env=None        )                stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))        self.stdio, self.write = stdio_transport        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))                await self.session.initialize()                # List available tools        response = await self.session.list_tools()        tools = response.tools        print("\nConnected to server with tools:", [tool.name for tool in tools])    async def process_query(self, query: str) -> str:        """Process a query using Claude and available tools"""        messages = [            {                "role": "user",                "content": query            }        ]        response = await self.session.list_tools()        available_tools = [{             "name": tool.name,            "description": tool.description,            "input_schema": tool.inputSchema        } for tool in response.tools]        # Initial Claude API call        response = self.anthropic.messages.create(            model="claude-3-5-sonnet-20241022",            max_tokens=1000,            messages=messages,            tools=available_tools        )        # Process response and handle tool calls        tool_results = []        final_text = []        for content in response.content:            if content.type == 'text':                final_text.append(content.text)            elif content.type == 'tool_use':                tool_name = content.name                tool_args = content.input                                # Execute tool call                result = await self.session.call_tool(tool_name, tool_args)                tool_results.append({"call": tool_name, "result": result})                final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")                # Continue conversation with tool results                if hasattr(content, 'text') and content.text:                    messages.append({                      "role": "assistant",                      "content": content.text                    })                messages.append({                    "role": "user",                     "content": result.content                })                # Get next response from Claude                response = self.anthropic.messages.create(                    model="claude-3-5-sonnet-20241022",                    max_tokens=1000,                    messages=messages,                )                final_text.append(response.content[0].text)        return "\n".join(final_text)    async def chat_loop(self):        """Run an interactive chat loop"""        print("\nMCP Client Started!")        print("Type your queries or 'quit' to exit.")                while True:            try:                query = input("\nQuery: ").strip()                                if query.lower() == 'quit':                    break                                    response = await self.process_query(query)                print("\n" + response)                                except Exception as e:                print(f"\nError: {str(e)}")        async def cleanup(self):        """Clean up resources"""        await self.exit_stack.aclose()async def main():    if len(sys.argv) < 2:        print("Usage: python client.py <path_to_server_script>")        sys.exit(1)            client = MCPClient()    try:        await client.connect_to_server(sys.argv[1])        await client.chat_loop()    finally:        await client.cleanup()if __name__ == "__main__":    import sys    asyncio.run(main())

代码解释

代码主体是MCP client的类，这代码实际上实现了host与client的简略功能

类的属性有：

        self.session: Optional[ClientSession] = None        self.exit_stack = AsyncExitStack()        self.anthropic = Anthropic()

exit_stack用于记录已有的通信(transport客户端)客户端通过session进行预设的方法调用以确定server提供的内容anthropic是连接claude模型的客户端，这里可以替换为其他任何可以进行tool calling的大模型。

类的方法有：

connect_to_server：主要按照MCP中的方法进行客户端的创建，这里可依据具体的业务场景，将服务发现的逻辑添加进去process_query是供人/host调用的接口，这里可以实现模型调用替换、提示词、tools选择逻辑甚至是host的功能调用。

通过执行uv run client.py <pyfile path>

附录

完整的server：

from typing import Anyimport httpxfrom mcp.server.fastmcp import FastMCPimport asyncioimport xml.etree.ElementTree as ETWEATHER_API_BASE = "https://restapi.amap.com/v3/weather/weatherInfo?parameters"USER_KEY="XXXXXXXXXXXXXX"mcp = FastMCP("china_weather")        def parse_city_xml(xml_string: str) -> dict:    try:        # 解析XML字符串        root = ET.fromstring(xml_string)        # 找到第一个<geocode>节点        geocode = root.find(".//geocode")        if geocode is None:            print("No geocode found in XML")            return {}                # 提取adcode和location        adcode = geocode.find("adcode").text if geocode.find("adcode") is not None else None        location = geocode.find("location").text if geocode.find("location") is not None else None                return {"adcode": adcode, "location": location}    except ET.ParseError as e:        print(f"XML parsing error: {e}")        return {}    except Exception as e:        print(f"Error processing XML: {e}")        return {}@mcp.tool()async def get_city_weather(adcode: str) -> str:    """Get today and 3day weather forecast for a adcode.    Args:        adcode: adcode or city name (e.g. "101010100" for Beijing, or "北京")    """    headers = {        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",        "Accept": "application/json"    }    params = {        "city": adcode,        "key": USER_KEY,        "extensions":"all"    }    url = "https://restapi.amap.com/v3/weather/weatherInfo"    async with httpx.AsyncClient() as client:        try:            response = await client.get(url, headers=headers, params=params, timeout=30.0)            response.raise_for_status()            data = response.json()        except Exception as e:            print(f"Request error: {e}")            return None        if not data or "forecasts" not in data:        return "无法获取该城市的天气信息或位置信息不正确"        forecast = data["forecasts"][0]  # Get 3day's forecast    casts = data["forecasts"][0]["casts"]        weather_info=""    for cast in casts:        forecast_str = (            f"日期: {cast.get('date', '未知')};"            f"白天天气: {cast.get('dayweather', '未知')};"            f"夜间天气: {cast.get('nightweather', '未知')};"            f"温度: {cast.get('nighttemp', '未知')}°C ~ {cast.get('daytemp', '未知')}°C;"            f"白天风向: {cast.get('daywind', '未知')};"            f"白天风力: {cast.get('daypower', '未知')}级"        )        weather_info+=forecast_str        return weather_info@mcp.tool()async def search_city_code(city_name: str) -> str:    """Search for location information by city name using Amap API.    Args:        city_name: Chinese city name or address (e.g. "北京市朝阳区阜通东大街6号")    """    params = {        "address": city_name,        "output": "XML",        "key": USER_KEY  # Replace with actual Amap API key    }    url = "https://restapi.amap.com/v3/geocode/geo"    headers = {        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",        "Accept": "application/xml"    }    async with httpx.AsyncClient() as client:        try:            response = await client.get(url, headers=headers, params=params, timeout=30.0)            response.raise_for_status()            xml_string=response.text        except httpx.HTTPStatusError as e:            print(f"HTTP error: {e.response.status_code} - {e}")            print(f"Request URL: {url}")            return None        except Exception as e:            print(f"Request error: {type(e).__name__} - {e}")            print(f"Request URL: {url}")            return None    data = parse_city_xml(xml_string)    location = {        "name": city_name,        "adcode": data["adcode"] if data["adcode"]  is not None else "未知",        "location": data["location"]  if data["location"]  is not None else "未知"    }    return f"地址: {location['name']}, 行政区划代码: {location['adcode']}, 经纬度: {location['location']}"if __name__ == "__main__":    # asyncio.run(mcp.run(transport="stdio"))    mcp.run(transport='stdio')

resource、Prompts、sampling的使用以及客户端的服务发现先鸽一下。