本文介绍了如何通过 LangChain 实现 MCP 调用。通过模拟一个简单的算术计算器，基于 MCP Server 运行，并使用 MCP Client 进行调用。最终，通过集成 DeepSeek 大模型完成整个 MCP 调用流程，同时提供了 Python 示例代码以供参考。

MCP是什么

1、MCP定义

Model Context Protocol（MCP）模型上下文协议是一种标准化协议，它让大模型能够更容易地和外部的数据、工具连接起来。你可以把MCP想象成一个通用的插头或者接口，就像USB-C一样，不管是什么设备，只要插上这个接口，就能和电脑、充电器等连接起来。

注意，它连接的不是物理设备，而是AI模型和外部的数据源、工具等。有了MCP，AI模型就能更方便地获取外部的信息，完成更多的任务。比如，通过MCP，AI模型可以操作电脑读写文件，或者模拟浏览器操作等。

2、MCP架构

• 客户端/服务器层：McpClient负责处理客户端操作，而McpServer则管理服务器端协议操作。两者都利用McpSession来进行通信管理。
• MCP服务器是模型上下文协议（MCP）架构中的基础组件，它为客户端提供工具、资源和功能。它实现了协议的服务器端部分。
• MCP客户端是模型上下文协议（MCP）架构中的关键组件，负责建立和管理与MCP服务器的连接。它实现了协议的客户端部分。
• 会话层（McpSession）：通过DefaultMcpSession实现来管理通信模式和状态。
• 传输层（McpTransport）：处理JSON-RPC消息的序列化和反序列化，并支持多种传输实现，比如Stdio、SSE等。
• SSE传输：客户端请求一个SSE（Server-Sent Events）通道以从服务器接收事件，然后通过HTTP POST请求发送命令。这种方式适用于需要跨网络通信的场景，通常用于分布式系统或需要高并发的场景。
• Stdio传输：客户端可以将MCP服务器作为本地子进程运行，并通过标准输入/输出直接与其通信。这种方式适用于本地集成和命令行工具，适合简单的本地批处理任务。

3、为什么需要MCP

首先，MCP提供了一个标准化的接口，使得AI模型能够轻松地与各种外部工具和数据源进行交互，无需为每个工具或数据源单独开发集成代码。

其次，MCP还解决了数据孤岛问题，通过统一协议连接分散的数据源，使AI模型能够实时访问和利用最新的数据。

总的来说，MCP就像是一个桥梁，让AI模型与外部世界更好地连接起来，从而发挥出更大的价值和潜力。

二、前提条件

1、Python运行环境安装。建议使用Python3.10以上版本，本示例使用了 Python 3.12.9版本。

2、Python开发工具安装。本人是vue\Java\python多种语言开发，所以使用了 IntelliJ IDEA开发工具。读者可以根据个人习惯选择合适的Python开发工具，比如：PyCharm、VS Code。

3、注册DeepSeek，获得api_key。访问deepseek的AI开放平台完成注册：https://platform.deepseek.com。

三、代码实现

创建python工程

首先，通过开发工具创建一个python工程。这一步很简单，不再描述。

接着，激活虚拟环境。在项目根目录下，并通过以下命令创建并激活虚拟环境：

python -m venv venv #有的环境下命令是python3或者py

.\venv\Scripts\activate #windows下的激活命令

注意：如果是通过IDEA工具创建的Python工程，默认会创建并激活venv虚拟环境，就不再需要手动创建了。

pip安装依赖包

本示例使用langchain、LangGraph、langchain-mcp-adapters和DeepSeek，所以需要先安装依赖包。

在虚拟环境命令窗口执行：

pip install -U langchain langgraph

pip install -U langchain-mcp-adapters

pip install -U langchain-deepseek

开发MCP Server

通过Python开发工具，创建一个python文件，命名为math_server.py。源代码如下：

# math_server.py

from mcp.server.fastmcp import FastMCP

import logging

# 配置日志记录器

logging.basicConfig(

level=logging.INFO, # 设置日志级别为 INFO

format="%(asctime)s - %(levelname)s - %(message)s" # 日志格式

)

logger = logging.getLogger(__name__)

# 创建 FastMCP 实例

mcp = FastMCP("Math")

@mcp.tool()

def add(a: int, b: int) -> int:

"""Add two numbers"""

logger.info("The add method is called: a=%d, b=%d", a, b) # 记录加法调用日志

return a + b

@mcp.tool()

def multiply(a: int, b: int) -> int:

"""Multiply two numbers"""

logger.info("The multiply method is called: a=%d, b=%d", a, b) # 记录乘法调用日志

return a * b

if __name__ == "__main__":

logger.info("Start math server through MCP") # 记录服务启动日志

mcp.run(transport="stdio") # 启动服务并使用标准输入输出通信

开发MCP Client

通过Python开发工具，创建一个python文件，命名为math_client.py。源代码如下：

import asyncio

from mcp import ClientSession, StdioServerParameters

from mcp.client.stdio import stdio_client

from langchain_mcp_adapters.tools import load_mcp_tools

from langgraph.prebuilt import create_react_agent

from langchain_deepseek import ChatDeepSeek

# 初始化 DeepSeek 大模型客户端

llm = ChatDeepSeek(

model="deepseek-chat", # 指定 DeepSeek 的模型名称

api_key="sk-e508ba61639640848060a1a2c1ee7b17" # 替换为您自己的 DeepSeek API 密钥

)

# 解析并输出结果

def print_optimized_result(agent_response):

"""

解析代理响应并输出优化后的结果。

:param agent_response: 代理返回的完整响应

"""

messages = agent_response.get("messages", [])

steps = [] # 用于记录计算步骤

final_answer = None # 最终答案

for message in messages:

if hasattr(message, "additional_kwargs") and "tool_calls" in message.additional_kwargs:

# 提取工具调用信息

tool_calls = message.additional_kwargs["tool_calls"]

for tool_call in tool_calls:

tool_name = tool_call["function"]["name"]

tool_args = tool_call["function"]["arguments"]

steps.append(f"调用工具: {tool_name}({tool_args})")

elif message.type == "tool":

# 提取工具执行结果

tool_name = message.name

tool_result = message.content

steps.append(f"{tool_name} 的结果是: {tool_result}")

elif message.type == "ai":

# 提取最终答案

final_answer = message.content

# 打印优化后的结果

print("\n计算过程:")

for step in steps:

print(f"- {step}")

if final_answer:

print(f"\n最终答案: {final_answer}")

# 定义异步主函数

async def main():

# 创建服务器参数

server_params = StdioServerParameters(

command="python",

# 确保更新为 math_server.py 文件路径

args=["./math_server.py"],

)

# 使用 stdio_client 进行连接

async with stdio_client(server_params) as (read, write):

async with ClientSession(read, write) as session:

# 初始化连接

await session.initialize()

print("成功连接到 Math 服务")

# 加载工具

tools = await load_mcp_tools(session)

print("加载工具完成: ", [tool.name for tool in tools])

# 创建代理

agent = create_react_agent(llm, tools)

# 循环接收用户输入

while True:

try:

# 提示用户输入问题

user_input = input("\n请输入您的问题（或输入 'exit' 退出）：")

if user_input.lower() == "exit":

print("感谢使用！再见！")

break

# 调用代理处理问题

agent_response = await agent.ainvoke({"messages": user_input})

# 打印完整响应（调试用）

print("\n完整响应:", agent_response)

# 调用抽取的方法处理输出结果

print_optimized_result(agent_response)

except Exception as e:

print(f"发生错误：{e}")

continue

# 使用 asyncio 运行异步主函数

if __name__ == "__main__":

asyncio.run(main())

四、运行测试

先运行math_server.py，再运行math_client.py，进行AI对话,观察日志输出结果，确定是否调用了自定义的MCP Server服务。

在线体验系统：http://www.yunchengxc.com