ai

1. 什么是 Function Calling？ #

Function Calling（函数调用）是大型语言模型（如 GPT、Claude）提供的一种能力：你可以事先告诉模型「有哪些函数可以调用」，模型在回答用户问题时，如果觉得需要用到某个函数，就会返回「我要调用某某函数，参数是……」的结构化信息；你的程序收到后，真正去执行这个函数，再把结果交给模型，由模型生成最终回答。

通俗理解：模型不会真的执行代码，它只会「决定要调用什么」。真正执行函数的是你的程序。就像顾问：顾问说「去查一下天气」，你（程序）去查，再把结果告诉顾问，顾问根据结果给出建议。

2. AI 模型如何「知道」有哪些函数？ #

模型本身不会自动知道你的业务函数。你需要用结构化描述告诉它，一般包括：

函数名：如 get_weather
功能说明：这个函数做什么，模型根据说明决定何时调用
参数：参数名、类型、是否必填、说明

模型根据这些描述，在合适的时候返回「调用请求」（函数名 + 参数值），由你的代码去真正执行。

3. Function Calling 的完整流程 #

定义函数：在代码里实现业务逻辑（如查天气、算加法）
描述函数：把函数名、说明、参数 schema 发给模型（通常在对话开始时或每次请求时）
用户提问：用户问「北京今天天气怎么样？」
模型决策：模型判断需要查天气，返回 {"name": "get_weather", "arguments": {"city": "北京"}}
执行函数：你的程序解析这个结构，调用 get_weather(city="北京")，得到结果
回传结果：把结果再发给模型
生成回答：模型根据结果生成自然语言回复给用户

4. 什么是 MCP？ #

MCP（Model Context Protocol，模型上下文协议）是由 Anthropic 提出的开放协议，用来统一 AI 应用与外部工具、数据源之间的交互方式。

核心思想：不依赖某一家模型厂商的私有 API，而是定义一套通用协议。任何支持 MCP 的客户端（如 Cursor、Claude Desktop）都可以连接任何 MCP 服务器，自动发现并调用其中的工具，无需为每个模型单独写适配代码。

MCP 中的主要概念：

概念	说明
工具（Tool）	可被模型调用的函数，相当于 function calling 中的「函数」
资源（Resource）	静态或动态数据，如文件、数据库记录
提示（Prompt）	可复用的提示模板

5. MCP 与 Function Calling #

在实际应用中，MCP 客户端通常作为「桥梁」，把 MCP 服务器的工具暴露给大模型，再把模型的调用请求转发给 MCP 服务器执行。完整流程如下：

5.1 流程概览 #

步骤	名称	说明
1	工具发现	MCP 客户端通过协议，从 MCP 服务器获取可用的工具列表（如 get_weather、send_email）
2	工具描述转换	客户端将 MCP 工具描述转换成 Function Calling 所需格式，传给大模型
3	模型决策	大模型根据用户提问，决定调用哪个工具，并通过 Function Calling 返回调用请求
4	请求路由	客户端收到调用请求后，通过 MCP 协议转发给对应的 MCP 服务器执行
5	结果返回	MCP 服务器执行完毕，通过 MCP 返回结果，客户端再通过 Function Calling 流程把结果交给模型，生成自然语言回答

5.2 流程示意 #

sequenceDiagram participant User as 用户 participant Client as 客户端<br/>(含工具描述) participant LLM as 大模型 participant MCP as MCP 服务器 User->>Client: 提问 Client->>LLM: 发送消息（携带工具描述） LLM-->>Client: 决策并返回工具调用请求 Client->>MCP: 转发工具调用请求 MCP-->>Client: 执行工具，返回结果 Client->>LLM: 把工具结果返回给大模型 LLM-->>Client: 生成并返回最终回答 Client-->>User: 输出答案

Function Calling 是模型层面的能力：模型厂商在 API 里提供「描述函数 + 返回调用请求」的机制，各家格式可能不同。
MCP 是协议层面的规范：把「工具」的声明、发现、调用、结果返回都标准化，任何兼容 MCP 的模型/客户端都能用同一套工具。

5.4 关系总结 #

MCP 标准化了 Function Calling：工具用统一格式描述，调用用统一协议，不再依赖某一家 API。
MCP 中的「工具」就是 Function Calling 的体现：模型通过 MCP 发现工具、发出调用请求，服务器执行并返回结果。
可以一起用：用 MCP 暴露工具，支持 MCP 的客户端（包括支持 function calling 的模型）都能直接使用，一次开发，多处复用。

5.5. 两者对比 #

维度	Function Calling	MCP
定位	模型 API 提供的能力	跨模型的通用协议
工具定义	各厂商格式不同	统一格式
适用场景	对接单一模型（如只用 OpenAI）	多模型、多客户端共用同一套工具
扩展性	主要围绕函数调用	工具 + 资源 + 提示

简单记：Function Calling 是「模型会调用函数」的能力；MCP 是「让所有模型都能用同一套工具」的协议。用 MCP 暴露工具，可以一次开发、多处复用。

1. 配置日志 #

在 MCP + Function Calling 的联调过程中，客户端、MCP 子进程、大模型 API 会交替输出大量信息。若缺少统一的日志配置，很难判断「请求卡在哪一步、工具是否被正确转发、模型返回了什么」。

本节引入可配置的日志基础设施，为后续打基础：

组件	作用
`Config`	从环境变量 `LOG_LEVEL` 读取日志等级，默认 `INFO`
`logger`	模块级日志记录器，供各函数引用
`setup_logging()`	统一设置输出格式、时间样式和日志等级
`__main__` 入口	启动时先加载配置，再初始化日志

日志等级可通过环境变量灵活切换，例如调试时设为 DEBUG：

set LOG_LEVEL=DEBUG
python main.py

时序图

sequenceDiagram participant U as 用户 / Shell participant M as main.py participant C as Config participant E as os.environ participant L as logging 模块 U->>M: python main.py M->>C: Config() C->>E: get("LOG_LEVEL", "INFO") E-->>C: 日志等级（如 INFO） C-->>M: config 对象 M->>M: setup_logging(config.log_level) M->>L: basicConfig(level, format, datefmt) L-->>M: 日志系统就绪 Note over M,L: 此后 logger 可按统一格式输出

1.1. main.py #

main.py

# 导入操作系统相关库
+import os

# 导入日志处理库
+import logging

# 定义配置类，用于读取和保存环境变量中的配置信息
+class Config:
    # 初始化方法
+   def __init__(self):
        # 读取环境变量中的日志等级，如果没有则默认使用"INFO"
+       self.log_level = os.environ.get("LOG_LEVEL", "INFO")

# 获取当前模块的日志记录器
+logger = logging.getLogger(__name__)

# 定义日志设置函数，设置日志等级和输出格式
+def setup_logging(level="INFO"):
    # 使用logging.basicConfig函数配置日志输出
+   logging.basicConfig(
        # 设置日志输出等级
+       level=getattr(logging, level.upper(), logging.INFO),
        # 设置日志输出的格式
+       format="%(asctime)s [%(levelname)s] %(message)s",
        # 设置日志日期时间的显示格式
+       datefmt="%H:%M:%S",
+   )

# 判断当前是否为主程序入口
+if __name__ == "__main__":
    # 创建配置对象，读取配置信息
+   config = Config()
    # 根据信息设置日志等级和格式
+   setup_logging(config.log_level)

2. MCP 客户端和服务器 #

本节把 main.py 扩展为同一文件、两种角色：既可以是 MCP 服务器，也可以是启动该服务器的客户端。这是 Function Calling 桥接的典型结构——客户端通过 stdio 拉起子进程中的 MCP 服务器，再从中发现可用工具。

运行方式	命令	角色
客户端（默认）	`python main.py`	启动 MCP 子进程，握手并列出工具
服务器	`python main.py serve`	以 stdio 模式对外暴露工具

本节新增的关键组件：

组件	作用
`.env`	存放 `DEEPSEEK_API_KEY`，供后续调用大模型时使用
`Config.get_mcp_server_params()`	构造子进程启动参数（`python main.py serve`）
`create_mcp_server()`	用 FastMCP 注册 `get_weather`、`send_email`、`add` 三个工具
`run_server()`	以 stdio 传输方式运行 MCP 服务器
`run_bridge()`	客户端入口：拉起子进程 → 初始化 → 发现工具
`MCPBridge`	桥接类骨架，后续章节会接入大模型

时序图

以默认方式 python main.py 运行为例，客户端与 MCP 子进程的交互如下：

sequenceDiagram participant U as 用户 / Shell participant C as main.py（客户端） participant SP as stdio_client participant S as main.py serve（子进程） participant MCP as FastMCP 服务器 U->>C: python main.py C->>C: Config() + setup_logging() C->>SP: stdio_client(server_params) SP->>S: 启动子进程 python main.py serve S->>MCP: run_server() / mcp.run(stdio) SP-->>C: read_stream, write_stream C->>S: ① initialize 请求 S-->>C: ② initialize 响应 C->>S: ③ notifications/initialized C->>S: tools/list S-->>C: get_weather, send_email, add C->>C: logger.info("工具发现: [...]") Note over C,MCP: 本节止于工具发现；下一节接入大模型

2.1. .env #

.env

DEEPSEEK_API_KEY=sk-ae8009b3b2f540d99f1cfa6ba7b3bd4d

2.2. main.py #

main.py

# 导入操作系统相关库
import os

# 导入日志处理库
import logging
# 导入系统参数和函数库
+import sys
# 导入异步相关库
+import asyncio
# 导入 OpenAI 认证异常
+from openai import AuthenticationError
# 导入 MCP 桥接通信相关类
+from mcp import ClientSession, StdioServerParameters
# 导入 MCP stdio 客户端工具
+from mcp.client.stdio import stdio_client
# 导入快速 MCP 服务器类
+from mcp.server.fastmcp import FastMCP
# 定义配置类，用于读取和保存环境变量中的配置信息
class Config:
    # 初始化方法
    def __init__(self):
        # 读取环境变量中的日志等级，如果没有则默认使用"INFO"
        self.log_level = os.environ.get("LOG_LEVEL", "INFO")
         # 读取 MCP 命令（默认是 python）
+       self.mcp_command = os.environ.get("MCP_COMMAND", "python")
        # 获取 MCP 服务器的启动参数
+   def get_mcp_server_params(self):
        # 构造 MCP 服务器启动参数
+      args = [__file__, "serve"]
        # 返回 MCP 服务器启动参数
+      return StdioServerParameters(command=self.mcp_command, args=args)    

# 获取当前模块的日志记录器
logger = logging.getLogger(__name__)

# 定义日志设置函数，设置日志等级和输出格式
def setup_logging(level="INFO"):
    # 使用logging.basicConfig函数配置日志输出
    logging.basicConfig(
        # 设置日志输出等级
        level=getattr(logging, level.upper(), logging.INFO),
        # 设置日志输出的格式
        format="%(asctime)s [%(levelname)s] %(message)s",
        # 设置日志日期时间的显示格式
        datefmt="%H:%M:%S",
    )
# MCP Bridge 类，负责连接 MCP 工具和大模型
+class MCPBridge:
+  """
+  连接 MCP 服务器与大模型，负责：
+  - 从 MCP 获取工具列表并转成 Function Calling 格式
+  - 调用大模型
+  - 将模型返回的工具调用请求转发给 MCP 执行
+  """      
    # 初始化 MCPBridge （保存配置和对象句柄）
+  def __init__(self, config):
+      self.config = config
# 异步函数，桥接主入口：调用 MCP 子进程并与大模型对话
+async def run_bridge(user_message, config):
    # 获取 MCP 服务器启动参数
+  server_params = config.get_mcp_server_params()
    # 使用 stdio_client 启动 MCP 服务器并建立通信流
+  async with stdio_client(server_params) as (read_stream, write_stream):
        # 建立 MCP 会话
+      async with ClientSession(read_stream, write_stream) as session:
            # 初始化会话
+          await session.initialize()
             # 获取已注册的工具
+          tools = await session.list_tools()
            # 日志记录已发现工具
+          logger.info("工具发现: %s", [t.name for t in tools.tools])
# 创建并注册工具的 MCP 服务器
+def create_mcp_server():
    # 实例化 FastMCP，命名为 MCP-Bridge
+  mcp = FastMCP(name="MCP-Bridge")

    # 注册获取天气工具
+  @mcp.tool()
+  def get_weather(city):
        # 查询指定城市的天气
+      return f"{city}今天晴，气温 25℃"

    # 注册发送邮件工具
+  @mcp.tool()
+  def send_email(to, subject, body):
        # 发送邮件到指定收件人
+      return f"已发送邮件给 {to}，主题：{subject}"

    # 注册加法计算工具
+  @mcp.tool()
+  def add(a, b):
        # 计算两个整数的和
+      return a + b

    # 返回 mcp 服务器对象
+  return mcp

# 创建 MCP 服务器实例
+mcp = create_mcp_server()            
# 以 stdio 模式运行 MCP 服务器，供子进程或编辑器调用
+def run_server():
    # 输出日志信息：
+  logger.info("MCP 服务器已启动（stdio 模式）")
    # 运行 MCP 服务（stdio 传输方式）
+  mcp.run(transport="stdio")
# 判断当前是否为主程序入口
if __name__ == "__main__":
    # 创建配置对象，读取配置信息
    config = Config()
    # 根据信息设置日志等级和格式
    setup_logging(config.log_level)
    # 判断参数是否要求以 serve 启动服务器
+   if len(sys.argv) >= 2 and sys.argv[1] == "serve":
        # 启动 stdio MCP 服务器
+      run_server()
+   else:
        # 否则当作命令行问答客户端
+      question = "北京今天天气怎么样？"
+      try:
            # 执行桥接对话
+          reply = asyncio.run(run_bridge(question, config))
            # 输出回复
+          print(reply)
        # 认证异常处理
+      except AuthenticationError:
+          logger.error("API 认证失败：请检查 DEEPSEEK_API_KEY 是否正确，可在 .env 中配置或设置环境变量")
+          sys.exit(1)

3. 将 MCP Tool 转为 OpenAI Function Calling 格式 #

MCP 与 OpenAI 对「工具」的描述格式不同。上一节已从 MCP 服务器发现了工具，但大模型 API 无法直接消费 MCP 的 inputSchema——需要先桥接转换，再作为 tools 参数传给模型。

两种格式的对应关系：

MCP Tool 字段	OpenAI Function 字段	说明
`name`	`function.name`	工具唯一标识
`description`	`function.description`	模型据此决定是否调用
`inputSchema` / `input_schema`	`function.parameters`	JSON Schema，描述参数类型与必填项

本节新增的核心逻辑：

组件	作用
`mcp_tools_to_openai_format()`	遍历 MCP 工具，提取 schema 并转为 OpenAI 的 `function` 定义
`MCPBridge.chat()`	拉取工具列表 → 转换格式 → 包装为 `[{"type": "function", ...}]`
`run_bridge()`	创建 `MCPBridge` 实例，初始化后调用 `chat()`

转换后的 tools 结构示例（模型 API 所需格式）：

[
  {
    "type": "function",
    "function": {
      "name": "get_weather",
      "description": "查询指定城市的天气",
      "parameters": {
        "type": "object",
        "properties": { "city": { "type": "string" } },
        "required": ["city"]
      }
    }
  }
]

若某工具缺少 schema，转换函数会填入空的 {"type": "object", "properties": {}, "required": []}，避免 API 报错。

时序图

以 python main.py 运行为例，工具发现与格式转换的流程如下：

sequenceDiagram participant U as 用户 / Shell participant RB as run_bridge() participant B as MCPBridge.chat() participant S as MCP 服务器 participant T as mcp_tools_to_openai_format() U->>RB: python main.py RB->>RB: 启动子进程 + session.initialize() RB->>B: chat(user_message, session) B->>S: list_tools() S-->>B: get_weather, send_email, add（含 inputSchema） B->>T: mcp_tools_to_openai_format(mcp_tools) T->>T: 提取 name / description / schema T-->>B: function_defs 列表 B->>B: 包装为 [{"type": "function", "function": ...}] B->>B: logger.info("tools: ...") Note over B,S: 本节止于格式转换与日志输出；下一节调用大模型 API

3.1. main.py #

main.py

# 导入操作系统相关库
import os

# 导入日志处理库
import logging

# 导入系统参数和函数库
import sys

# 导入异步相关库
import asyncio

# 导入 OpenAI 认证异常
from openai import AuthenticationError

# 导入 MCP 桥接通信相关类
from mcp import ClientSession, StdioServerParameters

# 导入 MCP stdio 客户端工具
from mcp.client.stdio import stdio_client

# 导入快速 MCP 服务器类
from mcp.server.fastmcp import FastMCP


# 定义配置类，用于读取和保存环境变量中的配置信息
class Config:
    # 初始化方法
    def __init__(self):
        # 读取环境变量中的日志等级，如果没有则默认使用"INFO"
        self.log_level = os.environ.get("LOG_LEVEL", "INFO")
        # 读取 MCP 命令（默认是 python）
        self.mcp_command = os.environ.get("MCP_COMMAND", "python")
        # 获取 MCP 服务器的启动参数

    def get_mcp_server_params(self):
        # 构造 MCP 服务器启动参数
+       args = [__file__, "serve"]
        # 返回 MCP 服务器启动参数
+       return StdioServerParameters(command=self.mcp_command, args=args)


# 获取当前模块的日志记录器
logger = logging.getLogger(__name__)


# 定义日志设置函数，设置日志等级和输出格式
def setup_logging(level="INFO"):
    # 使用logging.basicConfig函数配置日志输出
    logging.basicConfig(
        # 设置日志输出等级
        level=getattr(logging, level.upper(), logging.INFO),
        # 设置日志输出的格式
        format="%(asctime)s [%(levelname)s] %(message)s",
        # 设置日志日期时间的显示格式
        datefmt="%H:%M:%S",
    )


# 将 MCP Tool 转为 OpenAI Function Calling 格式
+def mcp_tools_to_openai_format(mcp_tools):
    # 定义保存结果的列表
+   result = []
    # 遍历每个工具
+   for tool in mcp_tools:
        # 获取工具的输入 schema（参数说明）
+       schema = (
+           getattr(tool, "inputSchema", None)
+           or getattr(tool, "input_schema", None)
+           or {}
+       )
        # 若 schema 为空，则使用默认格式
+       if not schema:
+           schema = {"type": "object", "properties": {}, "required": []}
        # 将该工具转为 openai 的 function call 格式并加入结果
+       result.append(
+           {
+               "name": tool.name,
+               "description": tool.description or f"调用工具 {tool.name}",
+               "parameters": schema,
+           }
+       )
    # 返回转换后的工具列表
+   return result


# MCP Bridge 类，负责连接 MCP 工具和大模型
class MCPBridge:
+   """
+   连接 MCP 服务器与大模型，负责：
+   - 从 MCP 获取工具列表并转成 Function Calling 格式
+   - 调用大模型
+   - 将模型返回的工具调用请求转发给 MCP 执行
+   """

    # 初始化 MCPBridge （保存配置和对象句柄）
+   def __init__(self, config):
+       self.config = config

    # 异步单轮/多轮对话，实现桥接逻辑
+   async def chat(self, user_message, session):
+       """
+       执行单轮对话：用户输入 -> 模型决策 -> 可选工具调用 -> 最终回答。
+       支持多轮工具调用（模型可能连续多次调用工具）。
+       """
        # 获取 MCP 工具列表
+       mcp_tools = list((await session.list_tools()).tools)
        # 转为 OpenAI function calling 格式
+       function_defs = mcp_tools_to_openai_format(mcp_tools)
        # 构造 tools 参数
+       tools = [{"type": "function", "function": f} for f in function_defs]
        # 构造 tools 参数
+       tools = [{"type": "function", "function": f} for f in function_defs]
+       logger.info("tools: %s", tools)


# 异步函数，桥接主入口：调用 MCP 子进程并与大模型对话
async def run_bridge(user_message, config):
    # 获取 MCP 服务器启动参数
+   server_params = config.get_mcp_server_params()
    # 生成 MCPBridge 实例
+   bridge = MCPBridge(config)
    # 使用 stdio_client 启动 MCP 服务器并建立通信流
+   async with stdio_client(server_params) as (read_stream, write_stream):
        # 建立 MCP 会话
+       async with ClientSession(read_stream, write_stream) as session:
            # 初始化会话
+           await session.initialize()
            # 获取已注册的工具
+           tools = await session.list_tools()
            # 日志记录已发现工具
+           logger.info("工具发现: %s", [t.name for t in tools.tools])
            # 调用桥接逻辑
+           return await bridge.chat(user_message, session)


# 创建并注册工具的 MCP 服务器
def create_mcp_server():
    # 实例化 FastMCP，命名为 MCP-Bridge
+   mcp = FastMCP(name="MCP-Bridge")

    # 注册获取天气工具
+   @mcp.tool()
+   def get_weather(city):
        # 查询指定城市的天气
+       return f"{city}今天晴，气温 25℃"

    # 注册发送邮件工具
+   @mcp.tool()
+   def send_email(to, subject, body):
        # 发送邮件到指定收件人
+       return f"已发送邮件给 {to}，主题：{subject}"

    # 注册加法计算工具
+   @mcp.tool()
+   def add(a, b):
        # 计算两个整数的和
+       return a + b

    # 返回 mcp 服务器对象
+   return mcp


# 创建 MCP 服务器实例
+mcp = create_mcp_server()


# 以 stdio 模式运行 MCP 服务器，供子进程或编辑器调用
def run_server():
    # 输出日志信息：
+   logger.info("MCP 服务器已启动（stdio 模式）")
    # 运行 MCP 服务（stdio 传输方式）
+   mcp.run(transport="stdio")


# 判断当前是否为主程序入口
if __name__ == "__main__":
    # 创建配置对象，读取配置信息
    config = Config()
    # 根据信息设置日志等级和格式
    setup_logging(config.log_level)
    # 判断参数是否要求以 serve 启动服务器
    if len(sys.argv) >= 2 and sys.argv[1] == "serve":
        # 启动 stdio MCP 服务器
+       run_server()
    else:
        # 否则当作命令行问答客户端
+       question = "北京今天天气怎么样？"
+       try:
            # 执行桥接对话
+           reply = asyncio.run(run_bridge(question, config))
            # 输出回复
+           print(reply)
        # 认证异常处理
+       except AuthenticationError:
+           logger.error(
+               "API 认证失败：请检查 DEEPSEEK_API_KEY 是否正确，可在 .env 中配置或设置环境变量"
+           )
+           sys.exit(1)

4. 接入大模型与工具调用循环 #

前三节完成了日志、MCP 连接和工具格式转换。本节把链路跑通：接入 DeepSeek API，实现「用户提问 → 模型决策 → MCP 执行工具 → 模型生成最终回答」的完整循环。

组件	作用
`load_dotenv()`	从 `.env` 加载 `DEEPSEEK_API_KEY` 等配置
`Config` 扩展	读取 `llm_api_key`、`llm_model`、`llm_base_url`
`AsyncOpenAI`	异步调用兼容 OpenAI 协议的大模型 API
`_message_to_dict()`	将模型返回的 message 对象转为可追加到历史的 dict
`MCPBridge.chat()`	多轮对话循环：调模型 → 执行工具 → 回填结果 → 再调模型

chat() 的核心是一个最多 5 轮的 while 循环（max_tool_rounds 防止模型反复调工具陷入死循环）：

步骤	动作
①	携带 `messages` + `tools` 调用大模型，`tool_choice="auto"`
②	若模型未返回 `tool_calls`，直接输出文本答案
③	若有 `tool_calls`，解析参数并转发给 MCP：`session.call_tool()`
④	将工具结果以 `role: tool` 追加到 `messages`，回到 ①

以默认问题「北京今天天气怎么样？」为例，模型会先调用 get_weather，拿到 MCP 返回的「北京今天晴，气温 25℃」后，再组织自然语言回复给用户。

时序图

sequenceDiagram participant U as 用户 participant B as MCPBridge.chat() participant LLM as DeepSeek API participant MCP as MCP 服务器 U->>B: "北京今天天气怎么样？" B->>MCP: list_tools() + 格式转换 B->>LLM: chat.completions.create(messages, tools) LLM-->>B: tool_calls: get_weather(city="北京") B->>MCP: call_tool("get_weather", {city: "北京"}) MCP-->>B: "北京今天晴，气温 25℃" B->>B: messages 追加 assistant + tool 结果 B->>LLM: chat.completions.create(更新后的 messages, tools) LLM-->>B: 自然语言最终回答（无 tool_calls） B-->>U: print(reply)

4.1. main.py #

main.py

# 导入操作系统相关库
import os

# 导入日志处理库
import logging

# 导入系统参数和函数库
import sys

# 导入异步相关库
import asyncio

# 导入 OpenAI 认证异常
from openai import AuthenticationError

# 导入 MCP 桥接通信相关类
from mcp import ClientSession, StdioServerParameters

# 导入 MCP stdio 客户端工具
from mcp.client.stdio import stdio_client

# 导入快速 MCP 服务器类
from mcp.server.fastmcp import FastMCP

# 导入 OpenAI 异步客户端
+from openai import AsyncOpenAI

# 导入 json 处理库
+import json
+from dotenv import load_dotenv

+load_dotenv(override=True)


# 定义配置类，用于读取和保存环境变量中的配置信息
class Config:
    # 初始化方法
    def __init__(self):
        # 读取环境变量中的日志等级，如果没有则默认使用"INFO"
        self.log_level = os.environ.get("LOG_LEVEL", "INFO")
        # 读取 MCP 命令（默认是 python）
        self.mcp_command = os.environ.get("MCP_COMMAND", "python")
        # 读取 LLM API Key
+       self.llm_api_key = os.environ.get(
+           "DEEPSEEK_API_KEY", "sk-ae8009b3b2f540d99f1cfa6ba7b3bd4d"
+       )
        # 读取 LLM 模型名称
+       self.llm_model = os.environ.get("LLM_MODEL", "deepseek-v4-pro")
        # 读取 LLM 基础 API 地址
+       self.llm_base_url = os.environ.get("LLM_BASE_URL", "https://api.deepseek.com")

    # 获取 MCP 服务器的启动参数
    def get_mcp_server_params(self):
        # 构造 MCP 服务器启动参数
        args = [__file__, "serve"]
        # 返回 MCP 服务器启动参数
        return StdioServerParameters(command=self.mcp_command, args=args)


# 获取当前模块的日志记录器
logger = logging.getLogger(__name__)


# 定义日志设置函数，设置日志等级和输出格式
def setup_logging(level="INFO"):
    # 使用logging.basicConfig函数配置日志输出
    logging.basicConfig(
        # 设置日志输出等级
        level=getattr(logging, level.upper(), logging.INFO),
        # 设置日志输出的格式
        format="%(asctime)s [%(levelname)s] %(message)s",
        # 设置日志日期时间的显示格式
        datefmt="%H:%M:%S",
    )


# 将 MCP Tool 转为 OpenAI Function Calling 格式
def mcp_tools_to_openai_format(mcp_tools):
    # 定义保存结果的列表
    result = []
    # 遍历每个工具
    for tool in mcp_tools:
        # 获取工具的输入 schema（参数说明）
        schema = (
            getattr(tool, "inputSchema", None)
            or getattr(tool, "input_schema", None)
            or {}
        )
        # 若 schema 为空，则使用默认格式
        if not schema:
            schema = {"type": "object", "properties": {}, "required": []}
        # 将该工具转为 openai 的 function call 格式并加入结果
        result.append(
            {
                "name": tool.name,
                "description": tool.description or f"调用工具 {tool.name}",
                "parameters": schema,
            }
        )
    # 返回转换后的工具列表
    return result


# 辅助函数：将 OpenAI 返回的 message 对象转为 API dict 格式
+def _message_to_dict(msg):
    # 初始化字典，角色为助手，内容取 message 的 content（若为 None 则设为 None）
+   d = {"role": "assistant", "content": msg.content or None}
    # 如果 message 包含工具调用（tool_calls）
+   if msg.tool_calls:
        # 如果有工具调用，则设置 content 为 None
+       d["content"] = None
        # 构建 tool_calls 列表，每个调用包含 id、类型、函数名和参数
+       d["tool_calls"] = [
+           {
+               "id": tc.id,
+               "type": "function",
+               "function": {
+                   "name": tc.function.name,
+                   "arguments": tc.function.arguments or "{}",
+               },
+           }
+           for tc in msg.tool_calls
+       ]
    # 返回处理后的字典
+   return d


# MCP Bridge 类，负责连接 MCP 工具和大模型
class MCPBridge:
    """
    连接 MCP 服务器与大模型，负责：
    - 从 MCP 获取工具列表并转成 Function Calling 格式
    - 调用大模型
    - 将模型返回的工具调用请求转发给 MCP 执行
    """

    # 初始化 MCPBridge （保存配置和对象句柄）
    def __init__(self, config):
        self.config = config
+       self._llm_client = AsyncOpenAI(
+           api_key=self.config.llm_api_key,
+           base_url=self.config.llm_base_url,
+       )

    # 异步单轮/多轮对话，实现桥接逻辑
    async def chat(self, user_message, session):
        """
        执行单轮对话：用户输入 -> 模型决策 -> 可选工具调用 -> 最终回答。
        支持多轮工具调用（模型可能连续多次调用工具）。
        """
        # 获取 MCP 工具列表
        mcp_tools = list((await session.list_tools()).tools)
        # 转为 OpenAI function calling 格式
        function_defs = mcp_tools_to_openai_format(mcp_tools)
        # 构造 tools 参数
        tools = [{"type": "function", "function": f} for f in function_defs]
        logger.info("tools: %s", tools)
        # 构造会话历史
+       messages = [
            # 系统消息：说明助手行为
+           {
+               "role": "system",
+               "content": "你是一个有帮助的助手。当用户需要查询天气、发邮件、计算时，请调用相应工具。",
+           },
            # 用户输入
+           {"role": "user", "content": user_message},
+       ]
        # 设置最多工具调用轮数，防止死循环
+       max_tool_rounds = 5  # 防止无限循环
+       round_count = 0
        # 循环进行多轮（最多5轮）交互
+       while round_count < max_tool_rounds:
            # 增加交互轮数
+           round_count += 1
            # 向大语言模型发送当前的会话消息和可用工具信息，获取模型的回复
+           response = await self._llm_client.chat.completions.create(
+               model=self.config.llm_model,  # 指定要使用的语言模型
+               messages=messages,  # 传递对话历史消息
+               tools=tools,  # 传递工具定义用于 Function Calling
+               tool_choice="auto",  # 让模型自动决定是否调用工具
+           )
            # 获取模型回复的 message
+           msg = response.choices[0].message
            # 转换为 openai API 需要的 dict 格式并添加到消息历史
+           messages.append(_message_to_dict(msg))
            # 当没有需要调用工具时，直接返回答案
+           if not msg.tool_calls:
+               return (msg.content or "").strip()
            # 执行模型要求的所有工具调用
+           for tc in msg.tool_calls:
                # 获取要调用的工具名称
+               name = tc.function.name
                # 获取工具调用参数字符串
+               args_str = tc.function.arguments or "{}"
+               try:
                    # 解析参数字符串为 dict
+                   arguments = json.loads(args_str)
+               except json.JSONDecodeError:
                    # 解析失败则使用空参数
+                   arguments = {}
                # 记录日志
+               logger.info("执行工具: %s", name)
                # 调用 MCP 工具
+               result = await session.call_tool(name, arguments=arguments)
                # 提取工具调用返回的文本内容
+               text = result.content[0].text if result.content else ""
                # 如果执行失败，记录错误信息
+               if result.isError:
+                   text = f"执行错误: {result.content}"
                # 将工具调用结果添加至消息历史
+               messages.append(
+                   {
+                       "role": "tool",
+                       "tool_call_id": tc.id,
+                       "content": text,
+                   }
+               )
        # 超过最大轮数未终止，返回提示信息
+       return "工具调用次数过多，已终止。"


# 异步函数，桥接主入口：调用 MCP 子进程并与大模型对话
async def run_bridge(user_message, config):
    # 获取 MCP 服务器启动参数
    server_params = config.get_mcp_server_params()
    # 生成 MCPBridge 实例
    bridge = MCPBridge(config)
    # 使用 stdio_client 启动 MCP 服务器并建立通信流
    async with stdio_client(server_params) as (read_stream, write_stream):
        # 建立 MCP 会话
        async with ClientSession(read_stream, write_stream) as session:
            # 初始化会话
            await session.initialize()
            # 获取已注册的工具
            tools = await session.list_tools()
            # 日志记录已发现工具
            logger.info("工具发现: %s", [t.name for t in tools.tools])
            # 调用桥接逻辑
            return await bridge.chat(user_message, session)


# 创建并注册工具的 MCP 服务器
def create_mcp_server():
    # 实例化 FastMCP，命名为 MCP-Bridge
    mcp = FastMCP(name="MCP-Bridge")

    # 注册获取天气工具
    @mcp.tool()
    def get_weather(city):
        # 查询指定城市的天气
        return f"{city}今天晴，气温 25℃"

    # 注册发送邮件工具
    @mcp.tool()
    def send_email(to, subject, body):
        # 发送邮件到指定收件人
        return f"已发送邮件给 {to}，主题：{subject}"

    # 注册加法计算工具
    @mcp.tool()
    def add(a, b):
        # 计算两个整数的和
        return a + b

    # 返回 mcp 服务器对象
    return mcp


# 创建 MCP 服务器实例
mcp = create_mcp_server()


# 以 stdio 模式运行 MCP 服务器，供子进程或编辑器调用
def run_server():
    # 输出日志信息：
    logger.info("MCP 服务器已启动（stdio 模式）")
    # 运行 MCP 服务（stdio 传输方式）
    mcp.run(transport="stdio")


# 判断当前是否为主程序入口
if __name__ == "__main__":
    # 创建配置对象，读取配置信息
    config = Config()
    # 根据信息设置日志等级和格式
    setup_logging(config.log_level)
    # 判断参数是否要求以 serve 启动服务器
    if len(sys.argv) >= 2 and sys.argv[1] == "serve":
        # 启动 stdio MCP 服务器
        run_server()
    else:
        # 否则当作命令行问答客户端
        question = "北京今天天气怎么样？"
        try:
            # 执行桥接对话
            reply = asyncio.run(run_bridge(question, config))
            # 输出回复
            print(reply)
        # 认证异常处理
        except AuthenticationError:
            logger.error(
                "API 认证失败：请检查 DEEPSEEK_API_KEY 是否正确，可在 .env 中配置或设置环境变量"
            )
            sys.exit(1)

导航菜单

1. 什么是 Function Calling？ #

2. AI 模型如何「知道」有哪些函数？ #

3. Function Calling 的完整流程 #

4. 什么是 MCP？ #

5. MCP 与 Function Calling #

5.1 流程概览 #

5.2 流程示意 #

5.4 关系总结 #

5.5. 两者对比 #

1. 配置日志 #

1.1. main.py #

2. MCP 客户端和服务器 #

2.1. .env #

2.2. main.py #

3. 将 MCP Tool 转为 OpenAI Function Calling 格式 #

3.1. main.py #

4. 接入大模型与工具调用循环 #

4.1. main.py #