随着Openai、Manus、字节、阿里等国内外AI厂商在Agent方向持续发力，调用外部工具成为Agent核心能力之一。

有别于 Function Calling 仅处理模型发起的单次函数调用，MCP 允许模型与工具在共享上下文中多轮协作，从而更好地完成复杂任务。

搞清楚MCP原理才能更灵活运用，刚接触MCP的同学可能都遇到过以下疑惑。

1. MCP 和 Function Calling 有什么区别和关系？

2. 模型是怎么知道用什么工具的？

3. MCP Client 和 MCP Host 什么关系？怎么开发落地？

一、什么是MCP？

MCP（模型上下文协议）是由 Anthropic 2024 年 11 月提出的一种通信协议，旨在提供一种让AI模型轻松访问控制外部工具的标准化方式。

MCP 由主机（Host）、客户端（Client）、服务器（Server）三大角色组成。

支持 stdio、streamable-http、sse等多种通信协议，用以适应不同模型厂商的部署环境。

MCP Host：用户前端入口，负责协调用户请求并调度模型。

MCP Client：与服务端建立通信通道，将 Host 发来的请求打包、转发给 Server

MCP Server：提供具体工具能力的模块

以高德MCP服务和本地文件管理MCP服务为例，直观感受下MCP的作用。

二、MCP vs Function Calling

Function Calling（函数调用）是AI应用服务商为实现工具调用而自定义的接口方式，不同AI服务商之间在接口定义和开发文档上存在差异。

MCP则是将复杂的工具函数调用抽象为客户端-服务器架构，像“USB-C接口”，定义了 LLM 与外部工具和数据源的通信格式。

以天气查询为例，AI应用Function Calling工具调用前，需要把工具信息定义好，写进配置。

而AI应用通过MCP进行工具调用前，Server端已经把工具信息封装好，应用只需接入Server地址即可获取可用的工具信息。

我们通过抓请求包，看下两种方式都做了哪些事情。

本次实验使用Client端http代理+Fiddler作为中间人截获请求信息。

代码地址：

https://github.com/era4d/ai-learning/tree/main/mcp_vs_function_call

也可采用Client端http代理+logger作为中间人，记录日志。

logger地址：

https://github.com/MarkTechStation/VideoCode/blob/main/MCP终极指南-番外篇/llm_logger.py

首先，启动天气查询服务

# weather_api_server.py
from fastapi import FastAPI, Queryfrom fastapi.responses import JSONResponseimport uvicorn
app = FastAPI()
@app.get("/weather")def get_weather(city: str = Query("纽约")):    # 只支持纽约，返回默认天气    if city.lower() in ["纽约", "new york"]:        return {"city": "New York", "weather": "Sunny", "temperature": "25°C"}    else:        return JSONResponse(content={"error": f"Only New York supported, got {city}"}, status_code=400)
if __name__ == "__main__":    uvicorn.run(app, host="0.0.0.0", port=5001)

Function Calling方式

用户询问“纽约天气怎么样？”​​​​​​​

# function_call_client.py
import requestsfrom langchain_openai import ChatOpenAIfrom langchain.tools import BaseToolfrom langgraph.prebuilt import create_react_agentimport asyncioimport httpx
# 代理开关，True表示启用代理，False表示禁用代理enable_proxy = False
class WeatherTool(BaseTool):    name: str = "weather"    description: str = "查询指定城市的天气（目前仅支持纽约）"
    def _run(self, city: str = "New York") -> str:        try:            resp = requests.get("http://localhost:5001/weather", params={"city": city})            data = resp.json()            if "error" in data:                return data["error"]            return f"{data['city']} 天气: {data['weather']}, 温度: {data['temperature']}"        except Exception as e:            return f"查询天气失败: {e}"
    async def _arun(self, city: str = "New York") -> str:        # 实现异步方法，实际上调用同步方法        return self._run(city)

async def main():    # 初始化天气工具    weather_tool = WeatherTool()
    proxy_url = "http://localhost:8888"
    # 使用上下文管理器创建异步httpx客户端，根据开关决定是否设置代理    client_kwargs = {"verify": False}    if enable_proxy:        client_kwargs["proxy"] = proxy_url        print("已启用代理:", proxy_url)    else:        print("未启用代理")
    async with httpx.AsyncClient(**client_kwargs) as http_client:
        # 初始化LLM（需配置OPENAI_API_KEY环境变量）        llm = ChatOpenAI(            api_key="sk-xxx",            model="qwen-max",            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",            http_async_client=http_client,            model_kwargs={                "tools": [                    {                        "type": "function",                         "function": {                            "name": "weather",                             "description": "查询指定城市的天气",                            "parameters": {                                "type": "object",                                "properties": {                                    "city": {                                        "type": "string",                                        "description": "城市名称，例如：New York"                                    }                                },                                "required": ["city"]                            }                        }                    }                ]            }        )
        # 打印模型信息        print("\n使用的模型:", llm.model_name)        # 创建ReAct代理        agent = create_react_agent(            llm,            tools=[weather_tool]        )
        weather_response = await agent.ainvoke(            {"messages": [{"role": "user", "content": "纽约天气怎么样？"}]}        )
        final_answer = weather_response["messages"][-1].content        print("代理返回结果:", final_answer)
# 启动异步主函数if __name__ == "__main__":    asyncio.run(main())

此时，应用服务端将用户问题和自身工具能力一并传给大模型。

这份完整版的大模型 AI 学习和面试资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

Request：应用服务发请求给大模型

大模型思考后，给出tool_calls调用weather工具的指令。

Response：大模型响应让服务端调用工具

应用端接收到大模型指令后执行了查纽约天气的任务，把结果返回给大模型。

Request：服务端调用完工具又把数据传给大模型

随后大模型把整理后的结果输出给用户。

Response：大模型输出给用户的最终结果

三、MCP方式

启动SSE模式的MCP服务器，通过@mcp.tool()快速将函数声明为 MCP 可调用的工具接口。​​​​​​​

# mcp_weather_server.py
from fastmcp import FastMCPimport uvicornimport requestsfrom typing import Optional
# 创建FastMCP应用实例mcp = FastMCP("Weather MCP Server")
@mcp.tool()def get_weather(city: str = "纽约") -> str:    """    查询指定城市的天气信息
    Args:        city: 城市名称，默认为纽约
    Returns:        天气信息字符串    """    try:        # 调用本地weather API服务        resp = requests.get("http://localhost:5001/weather", params={"city": city})        data = resp.json()
        if "error" in data:            return f"错误: {data['error']}"
        return f"{data['city']} 天气: {data['weather']}, 温度: {data['temperature']}"    except Exception as e:        return f"查询天气失败: {str(e)}"
@mcp.tool()def get_supported_cities() -> str:    """    获取支持的城市列表
    Returns:        支持的城市列表    """    return "目前仅支持查询纽约(New York)的天气信息"
if __name__ == "__main__":    # 启动SSE模式的MCP服务器    mcp.run(        transport="sse",  # 或 "http"        host="0.0.0.0",        port=8000,        path="/sse"    )

MCP Client端将server接入，通过get_tools()获取可用的工具并完成工具调用。​​​​​​​

# mcp_client.py
from langchain_mcp_adapters.client import MultiServerMCPClientfrom langgraph.prebuilt import create_react_agentfrom langchain_openai import ChatOpenAIimport asyncioimport httpx
# 代理开关，True表示启用代理，False表示禁用代理enable_proxy = True
async def main():    proxy_url = "http://localhost:8888"
    # 禁用SSL警告    import urllib3    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    # 为 MCP 客户端设置代理环境变量    import os    if enable_proxy:        os.environ["HTTP_PROXY"] = proxy_url        os.environ["HTTPS_PROXY"] = proxy_url        print("已启用代理:", proxy_url)    else:        # 清除代理环境变量        os.environ.pop("HTTP_PROXY", None)        os.environ.pop("HTTPS_PROXY", None)        print("未启用代理")
    # 使用上下文管理器创建异步httpx客户端，根据开关决定是否设置代理    client_kwargs = {"verify": False}    if enable_proxy:        client_kwargs["proxy"] = proxy_url
    async with httpx.AsyncClient(**client_kwargs) as http_client:        # 创建LLM实例        llm = ChatOpenAI(            api_key="sk-dce77e3fea4e4c75bc208ad33ca7e7eb",            model="qwen-max",            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",            http_async_client=http_client  # 传入带代理的异步httpx客户端        )

        client = MultiServerMCPClient(            {                "weather": {                    # 确保你在 8000 端口启动天气服务器                    "url": "http://localhost:8000/sse",                    "transport": "sse",                }            }        )
        tools = await client.get_tools()
        agent = create_react_agent(            llm,            tools        )
        weather_response = await agent.ainvoke(            {"messages": [{"role": "user", "content": "what is the weather in 纽约?"}]}        )
        final_answer = weather_response["messages"][-1].content        print(final_answer)

# 启动异步主函数if __name__ == "__main__":    asyncio.run(main())

MCP Client发送3条请求，通知Server初始化已完成，并通过"tools/list"向Server索要工具列表。

sse请求

拿到工具列表后的流程和Function Calling一样；

将用户问题和工具列表发给大模型，大模型分析后指导Host调用get_weather工具，最后将结果返回给用户。

Request：应用（Host）将用户问题及获取到的工具列发给大模型

Response：大模型响应，让Host调用get_weather

Request：调用工具并将结果给大模型

Response：大模型将结果整理后输出给用户

最后通过列表总结Function Calling和MCP的区别和关系。

四、开发实践

目前很多厂商优秀的Agent框架（如：Eino、JManus、Langchain等）都已集成MCP能力。

为了进一步理解MCP，并将其灵活应用到业务中，我们将采用更原生的方式基于官方python sdk写一个支持多mcp server的mcp client，并提供stdio、SSE、StreamableHTTP、WebSocket协议的通信能力。

项目地址：https://github.com/era4d/mcp-client-python

项目结构：​​​​​​​

项目根目录├── client.py                # 程序主入口，负责加载配置并连接各MCP服务器├── servers.yaml             # 服务器配置文件，定义各MCP Server的连接方式和参数├── .env                     # 环境变量配置，如API Key等├── pyproject.toml           # Python项目依赖与元数据├── core/                    # 核心功能模块│   ├── mcp_client.py        # MCP客户端核心逻辑，管理会话、工具调用、与服务器通信│   ├── context_manager.py   # 上下文管理器，负责对话历史、工具调用记录等│   ├── llm_service.py       # LLM服务封装，负责与大模型API交互│   └── logger.py            # 日志系统初始化与管理├── servers/                 # 示例MCP服务器实现│   ├── calc.py              # 计算器服务│   ├── crawler.py           # 网络爬虫服务│   ├── weather.py           # 天气查询服务│   └── wiki.py              # 安全咨询查询服务├── logs/                    # 日志与上下文历史存储│   ├── context_history.json # 对话历史记录│   └── mcp_client.log       # 运行日志├── test/                    # 测试用例│   └── test_context.py      # 上下文管理器测试└── .gitignore               # Git忽略文件配置

五、MCP Server

MCP现支持C#、Go、Java、TypeScript等多种语言的SDK。

项目示例采用python SDK，并提供4种不同通信协议的mcp server样例。

同时可以通过以下Prompt快速生成mcp server。​​​​​​​

# 需求基于 MCP 相关资料，建一个 MCP Server，需求如下：
- 提供一个查询天气的工具- 采用sse通信协议- 要求功能简洁、只包含关键功能- 使用 Python 编写
# 请访问链接获取MCP server 开发参考资料：https://modelcontextprotocol.io/quickstart/server

MCP Debugging

可以通过MCP Inspector对MCP server进行调试跟踪。​​​​​​​

# Terminal 运行npx @modelcontextprotocol/inspector

MCP Inspector

MCP Client

client端主要包括server连接初始化、process_query处理用户输入并调用工具、cleanup关闭连接释放资源。

from mcp.client.stdio import stdio_clientfrom mcp.client.sse import sse_clientfrom mcp.client.streamable_http import streamablehttp_clientfrom mcp.client.websocket import websocket_client
class MCPClient:    ...    async def _connect_stdio(self, server: dict):        """连接 stdio 模式 Server"""        try:            params = StdioServerParameters(                command=server.get("command", "python"),                args=[server["path"]],                env=None            )            logger.info(f"正在连接stdio服务器: {server.get('name')}")            # 添加超时机制，避免无限等待            return await asyncio.wait_for(                self.exit_stack.enter_async_context(stdio_client(params)),                timeout=10.0  # 10秒超时            )        except asyncio.TimeoutError:            logger.error(f"连接stdio服务器超时: {server.get('name')}")            raise        except Exception as e:            logger.error(f"连接stdio服务器失败: {server.get('name')}, 错误: {e}")            import traceback            logger.error(traceback.format_exc())            raise
    async def _connect_sse(self, server: dict):        """连接 SSE 模式 Server"""        url = server.get("url")        if not url:            raise ValueError(f"SSE服务器 {server.get('name')} 未提供URL")
        logger.info(f"正在连接SSE服务器: {url}")        try:            return await self.exit_stack.enter_async_context(sse_client(url))        except Exception as e:            logger.error(f"连接SSE服务器失败: {url}, 错误: {e}")            import traceback            logger.error(traceback.format_exc())            raise
    async def _connect_streamable_http(self, server: dict):        """连接 StreamableHTTP 模式 Server"""        url = server.get("url")        if not url:            raise ValueError(f"StreamableHTTP服务器 {server.get('name')} 未提供URL")
        headers = server.get("headers", {})        logger.info(f"正在连接StreamableHTTP服务器: {url}")        try:            result = await self.exit_stack.enter_async_context(streamablehttp_client(url, headers=headers))            logger.debug(f"StreamableHTTP连接结果: {result}, 类型: {type(result)}, 长度: {len(result) if hasattr(result, '__len__') else 'N/A'}")            # streamablehttp_client返回(read_stream, write_stream, get_session_id_callback)            # 但ClientSession只需要前两个参数            if isinstance(result, tuple) and len(result) >= 2:                return (result[0], result[1])  # 只返回read_stream和write_stream            else:                raise ValueError(f"StreamableHTTP客户端返回了意外的结果格式: {type(result)}")        except Exception as e:            logger.error(f"连接StreamableHTTP服务器失败: {url}, 错误: {e}")            import traceback            logger.error(traceback.format_exc())            raise
    async def _connect_websocket(self, server: dict):        """连接 WebSocket 模式 Server"""
        url = server.get("url")        if not url:            raise ValueError(f"WebSocket服务器 {server.get('name')} 未提供URL")
        logger.info(f"正在连接WebSocket服务器: {url}")        try:            return await self.exit_stack.enter_async_context(websocket_client(url))        except Exception as e:            logger.error(f"连接WebSocket服务器失败: {url}, 错误: {e}")            import traceback            logger.error(traceback.format_exc())            raise

最后简单说说模型是怎么知道用什么工具的。
为了让大模型具有工具调用能力，模型在预训练或微调时，采用了数千条工具调用语料来训练模型，格式类似于下面的json数据。​​​​​​​

{    "conversations": [        {            "from": "human",            "value": "今天天气怎么样?"        },        {            "from": "gpt",            "value": "你想获得某个地方的天气情况，请提供地点信息"        },        {            "from": "human",            "value": "北京"        },        {            "from": "gpt",            "value": "{\n    \"function\": \"get_weather\",\n    \"arguments\": {\n        \"location\": \"北京\"\n    }\n}"        }    ]}

AI大模型从0到精通全套学习大礼包

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

只要你是真心想学AI大模型，我这份资料就可以无偿共享给你学习。大模型行业确实也需要更多的有志之士加入进来，我也真心希望帮助大家学好这门技术，如果日后有什么学习上的问题，欢迎找我交流，有技术上面的问题，我是很愿意去帮助大家的！

如果你也想通过学大模型技术去帮助就业和转行，可以点扫描下方链接👇👇
大模型重磅福利：入门进阶全套104G学习资源包免费分享！

01.从入门到精通的全套视频教程

包含提示词工程、RAG、Agent等技术点

​

02.AI大模型学习路线图（还有视频解说）

全过程AI大模型学习路线

​​

03.学习电子书籍和技术文档

市面上的大模型书籍确实太多了，这些是我精选出来的

04.大模型面试题目详解

05.这些资料真的有用吗?

这份资料由我和鲁为民博士共同整理，鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位，在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利，同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。

所有的视频由智泊AI老师录制，且资料与智泊AI共享，相互补充。这份学习大礼包应该算是现在最全面的大模型学习资料了。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

智泊AI始终秉持着“让每个人平等享受到优质教育资源”的育人理念‌，通过动态追踪大模型开发、数据标注伦理等前沿技术趋势‌，构建起"前沿课程+智能实训+精准就业"的高效培养体系。

课堂上不光教理论，还带着学员做了十多个真实项目。学员要亲自上手搞数据清洗、模型调优这些硬核操作，把课本知识变成真本事‌！

如果说你是以下人群中的其中一类，都可以来智泊AI学习人工智能，找到高薪工作，一次小小的“投资”换来的是终身受益！

应届毕业生‌：无工作经验但想要系统学习AI大模型技术，期待通过实战项目掌握核心技术。

零基础转型‌：非技术背景但关注AI应用场景，计划通过低代码工具实现“AI+行业”跨界‌。

业务赋能 ‌突破瓶颈：传统开发者（Java/前端等）学习Transformer架构与LangChain框架，向AI全栈工程师转型‌。

👉获取方式：
😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓