简介

MCP 是 Anthropic (Claude) 主导发布的一个开放的、通用的、有共识的协议标准。

• MCP 是一个标准协议，就像给 AI 大模型装了一个 “万能接口”，让 AI 模型能够与不同的数据源和工具进行无缝交互。它就像 USB-C 接口一样，提供了一种标准化的方法，将 AI 模型连接到各种数据源和工具。

• MCP 旨在替换碎片化的 Agent 代码集成，从而使 AI 系统更可靠，更有效。通过建立通用标准，服务商可以基于协议来推出它们自己服务的 AI 能力，从而支持开发者更快的构建更强大的 AI 应用。开发者也不需要重复造轮子，通过开源项目可以建立强大的 AI Agent 生态。

• MCP 可以在不同的应用 / 服务之间保持上下文，从而增强整体自主执行任务的能力。

MCP 遵循客户端 - 服务器架构，包含以下几个核心部分：

• MCP 主机（MCP Hosts）：发起请求的 AI 应用程序，比如聊天机器人、AI 驱动的 IDE 等。

• MCP 客户端（MCP Clients）：在主机程序内部，与 MCP 服务器保持 1:1 的连接。

• MCP 服务器（MCP Servers）：为 MCP 客户端提供上下文、工具和提示信息。

• 本地资源（Local Resources）：本地计算机中可供 MCP 服务器安全访问的资源，如文件、数据库。

• 远程资源（Remote Resources）：MCP 服务器可以连接到的远程资源，如通过 API 提供的数据。

基础概念

• 架构： 模型上下文协议(MCP)建立在灵活、可扩展的架构之上,可以实现 LLM 应用程序和集成之间的无缝通信。

• 资源： 资源是模型上下文协议(MCP)中的一个核心原语，允许服务器暴露数据和内容，这些内容可以被客户端读取并用作 LLM 交互的上下文。

• 工具： 工具是模型上下文协议(MCP)中的一个强大原语，使服务器能够向客户端暴露可执行的功能。通过工具，LLM 可以与外部系统交互、执行计算并在现实世界中采取行动。工具设计为模型控制，这意味着工具从服务器暴露给客户端时，目的是让 AI 模型能够自动调用它们（在人工审批的情况下）。

• 提示词： 提示词使服务器能够定义可重用的提示词模板和工作流,客户端可以轻松地将其呈现给用户和 LLM。它们提供了一种强大的方式来标准化和共享常见的 LLM 交互。提示词设计为用户控制,这意味着它们从服务器暴露给客户端时,用户可以明确地选择使用它们。

• 采样： 采样是 MCP 的一个强大功能，允许服务器通过客户端请求 LLM 补全，从而实现复杂的代理行为，同时保持安全性和隐私性。

• 传输： 模型上下文协议(MCP)中的传输为客户端和服务器之间的通信提供基础。传输层负责处理消息如何发送和接收的底层机制。

基本工作流程

1. 连接：MCP 主机连接到一个或多个 MCP 服务器。

2. 请求：主机发送请求以获取数据或执行工具。

3. 处理：服务器处理请求，访问相关数据源或外部服务。

4. 返回：服务器将结果返回给主机。

5. 生成响应：主机将信息提供给 AI 模型，用于生成用户响应。

入门实践

1. 环境配置

环境搭建：

1. pip install uv
2. uv init mcp-client #创建目录
3. cd mcp-client[添加链接描述](https://image2url.com/r2/default/images/1768201960725-6566e8aa-9848-4854-80c0-c914f263a390.jpeg)
4. uv venv
5. .venv\Scripts\activate # Windows系统
   source .venv/bin/activate # Unix或MacOS系统
6. uv add mcp anthropic python-dotenv # 安装必要文件

基础配置：

# .env 环境配置
BASE_URL=https://api.deepseek.com # 模型api地址
MODEL=deepseek-chat               # 模型名称     
OPENAI_API_KEY="your api key"     # api key

2. 简单的服务器搭建

from fastmcp import FastMCP
mcp = FastMCP()

@mcp.tool()
def get_weather(city:str):
    """
    获取对应城市的天气
    :param city: 城市
    :return: 城市天气的描述
    :type city: str
    """
    
    return "天气：晴，气温：20°"

if __name__ == '__main__':
    mcp.run()

@mcp.tool(): 将函数声明成 mcp tool，在函数内通过注解对函数的使用范围、输入输出进行描述，方便模型判断何时对其进行调用。（何时执行的逻辑并不在mcp服务器中）

3. 简单的客户端搭建

from fastmcp import Client
from openai import OpenAI
from dotenv import load_dotenv
from typing import List,Dict
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
 
import json
import asyncio
import os

load_dotenv()
class UserClient:
    def __init__(self,script="server.py"):
        self.mcp_client = Client(script)
        self.openai_client = OpenAI(
            base_url = "https://api.deepseek.com",
            api_key = "sk-0eef247c26094fe3a380715a86692dfd",
        )
        self.messages = [
            {
                "role" : "system",
                "content" : "你是一个 AI 助手，你需要借助工具回答用户的问题"
            }
        ]
        self.tools = []
    
    async def get_tools(self):
        tools = await self.mcp_client.list_tools()
        tools = [
            {
                "type":"function",
                "function": {
                    "name" : tool.name,
                    "description" : tool.description,
                    "input_schema" : tool.inputSchema
                }
            }
            for tool in tools
        ]
        return tools
    
    async def chat(self,messages:List[Dict]):
        # print("!!!")
        if not self.tools:
            self.tools = await self.get_tools()
        response = self.openai_client.chat.completions.create(
            model = "deepseek-chat" ,
            messages = messages,
            tools = self.tools
        )

        self.messages.append({
                "role": "assistant",
                "content": response.choices[0].message.content,
                "tool_calls": response.choices[0].message.tool_calls
            })
        
        # print(response)
        content = response.choices[0]
        if content.finish_reason != "tool_calls":
            return content.message

        for tool_call in content.message.tool_calls:
            final_response =await self.mcp_client.call_tool(
                tool_call.function.name,
                json.loads(tool_call.function.arguments)
            )
            # print(response.content[0].text)
            self.messages.append({
                "role": "tool",
                "content": final_response.content[0].text,
                "tool_call_id": tool_call.id
            })
        print(self.messages)
        return await self.chat(self.messages)



    async def loop(self):
        async with self.mcp_client:
            while True:
                question = input("User: ")
                messages = {
                    "role" : "user",
                    "content" : question
                }
                self.messages.append(messages)
                if(question == "exit" ) :
                    print("已退出天气查询助手")
                    break
                response_message = await self.chat(self.messages)
                print("AI: ",response_message.content)

async def main():
    user_client = UserClient()
    await user_client.loop()

if __name__ == '__main__':
    asyncio.run(main())

执行流程：

异步等待： 执行chat()、lopp()、call_tool()、get_tools() 时必须进行异步处理

MCP tool 工具调用后的上下文处理 ：必须在messages中加入response.choices[0].message.tool_calls和tool_call.id，否则模型会一直返会tool_calls，不停请求客户端执行工具的调用

• tool_calls: 大模型返回的工具调用信息，包含所需调用的function的名称和相关参数，必须作为历史信息加入到self.messages中

• tool_call.id：模型赋予工具的唯一标识符，在工具调用结束后需要将其和工具调用结果一同加入到历史信息中，使大模型知道客户端执行了哪个function