OpenAI Agent SDK + MCP协议:从零构建你的第一个AI工具调用服务

最近在折腾AI应用开发时,我发现一个挺有意思的现象:很多开发者已经能熟练调用大模型的对话接口,但一旦涉及到让AI去“动手”操作外部系统——比如查个天气、调个数据库,或者控制一下智能家居——整个流程就变得复杂起来。要么得写一堆繁琐的提示词去描述工具,要么得自己处理复杂的函数调用逻辑。直到我深入研究了OpenAI Agent SDK与MCP(Model Context Protocol)协议的组合,才真正找到了一条优雅的路径。

这套组合拳的核心思想,是将AI的“思考”能力与“执行”能力进行标准化解耦。你可以把MCP协议想象成一个万能适配器,它定义了一套清晰的规范,让任何外部工具(从简单的计算器到复杂的企业API)都能以一种AI模型能理解的方式被“描述”和“调用”。而OpenAI Agent SDK则扮演了智能调度中心的角色,它基于这套协议,自动管理工具的选择、参数的传递以及结果的解析。这意味着,作为开发者,你不再需要为每一个新工具去重写大量的胶水代码,只需按照MCP的标准“封装”好你的工具,剩下的“让AI决定何时用、怎么用”的工作,SDK几乎全包了。

这篇文章,就是为你——那些希望将AI能力无缝嵌入到具体业务逻辑、构建真正“能动起来”的智能应用的开发者——准备的一份实战指南。我们将从最基础的概念梳理开始,一步步搭建环境,亲手编写并连接MCP服务端与客户端,最终实现一个能自主调用多种工具的AI智能体。过程中,我会分享一些在官方文档里未必会提到的配置细节和调试技巧。

1. 理解核心基石:MCP协议与Agent SDK的角色分工

在动手写代码之前,花点时间弄清楚MCP和Agent SDK各自管什么、怎么配合,能让你在后续开发中少走很多弯路。这绝不是两个孤立的技术栈的简单叠加,而是一种精心设计的分层架构。

MCP协议:工具世界的“标准化说明书”

你可以把MCP理解为一套为AI模型与外部工具交互而设计的“通信协议”或“接口规范”。它的核心目标是解决一个根本问题:如何让一个原本只懂自然语言的AI模型,能够理解、发现并安全地调用千差万别的外部功能?

MCP通过几个关键抽象实现了这一点:

  • 工具(Tools):这是最核心的概念。任何你想让AI调用的功能,比如“查询天气”、“发送邮件”、“执行数据库查询”,都需要被定义为一个工具。每个工具必须有清晰的名称、描述和参数规范。
  • 资源(Resources):代表AI可以读取的静态或动态数据源,比如一个文本文件、一个数据库表的模式定义,或者一个API的实时状态。资源为AI提供了额外的上下文信息。
  • 提示(Prompts):可复用的对话模板或指令片段,帮助引导AI在特定场景下更好地使用工具和资源。

MCP协议规定了这些组件如何被“描述”(通常使用结构化的JSON Schema)以及如何被“调用”(通过标准的RPC,如SSE或stdio)。一个符合MCP标准的服务器,本质上就是一个对外暴露了上述标准化接口的服务。

OpenAI Agent SDK:智能体的“大脑”与“调度器”

如果说MCP定义了工具长什么样、怎么用,那么OpenAI Agent SDK就是那个决定“什么时候、用哪个工具、达到什么目的”的智能中枢。它基于OpenAI的模型(如GPT-4),内置了强大的推理和规划能力。

SDK中的 Agent 类是这个智能体的核心。你通过 instructions 赋予它角色和目标,通过 mcp_servers 参数为它装备上由MCP协议封装好的工具库。当用户提出一个请求时,Agent SDK内部会完成以下自动化流程:

  1. 理解意图:分析用户的自然语言输入。
  2. 工具发现与选择:自动从已连接的MCP服务器中检索可用的工具列表,并判断哪个或哪些工具最适合当前任务。
  3. 参数提取与验证:从用户输入中提取出工具所需的参数,并按照MCP定义的模式进行校验。
  4. 执行调用:通过MCP协议向对应的服务器发起工具调用请求。
  5. 结果处理与响应:接收工具返回的结果,将其整合到上下文中,并生成最终的自然语言回复给用户。

这个过程中,开发者几乎不需要手动干预工具调用的决策逻辑。这种分工带来的最大好处是灵活性与可维护性。你可以独立地开发、升级你的工具服务器(MCP Server),只要接口符合规范,智能体(Agent)就能即插即用,无需修改核心AI逻辑。

提示:可以把这种架构类比为“插件系统”。MCP Server是各式各样的插件,提供具体功能;Agent SDK是主程序,负责管理和调度这些插件来完成任务。

2. 搭建开发环境:从零开始的准备工作

理论清楚了,我们开始动手。一个清晰、隔离的Python环境是项目成功的起点。这里我强烈推荐使用 uv 这个新兴的、速度极快的Python包管理器和安装器,它能大幅提升依赖管理的效率。

首先,确保你的系统已经安装了Python 3.10或更高版本。然后,我们来一步步搭建项目骨架。

步骤一:初始化项目与虚拟环境

在你的项目目录下,打开终端,执行以下命令来创建虚拟环境。使用 uv 可以一步到位,它比传统的 venvpip 组合要快得多。

# 使用 uv 创建虚拟环境,环境目录命名为 .venv
uv venv .venv

# 激活虚拟环境 (Windows)
.\.venv\Scripts\activate
# 激活虚拟环境 (macOS/Linux)
source .venv/bin/activate

激活后,你的命令行提示符前应该会出现 (.venv) 字样,表示你已经进入了这个独立的Python环境。

步骤二:定义项目依赖

接下来,我们需要创建一个 pyproject.toml 文件来声明项目信息和依赖。这是现代Python项目的标准做法。将以下内容保存到项目根目录的 pyproject.toml 文件中。

[project]
name = "my-ai-agent-service"
version = "0.1.0"
description = "一个基于OpenAI Agent SDK和MCP协议的AI工具调用服务示例"
authors = [
    {name = "Your Name", email = "your.email@example.com"},
]
requires-python = ">=3.10"
dependencies = [
    "openai>=1.66.5",          # OpenAI官方Python SDK
    "openai-agents==0.0.7",    # OpenAI Agent SDK (请关注最新版本)
    "mcp>=1.0.0",              # MCP协议Python实现库
    "requests>=2.31.0",        # 用于HTTP请求,我们的天气工具会用到
    "python-dotenv>=1.0.0",    # 用于管理环境变量
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

步骤三:安装依赖

有了 pyproject.toml 文件,使用 uv 安装所有依赖就变得非常简单:

uv sync

这个命令会读取 pyproject.toml,快速下载并安装所有列出的包到你的虚拟环境中。相比 pip install -r requirements.txtuv sync 通常更快,并且能更好地处理依赖解析。

步骤四:配置API密钥

为了使用OpenAI的模型,你需要一个API Key。创建一个名为 .env 的文件(注意前面的点),将你的密钥放入其中。务必确保这个文件被添加到 .gitignore 中,避免密钥泄露。

# .env 文件内容
OPENAI_API_KEY=sk-your-actual-openai-api-key-here
# 如果你使用其他兼容OpenAI API的提供商,可以设置BASE_URL
# OPENAI_API_BASE=https://api.your-provider.com/v1

至此,一个干净、高效的开发环境就准备就绪了。接下来,我们将进入最核心的环节:构建我们自己的MCP工具服务器。

3. 构建你的第一个MCP服务器:封装自定义工具

MCP服务器是我们所有自定义功能的载体。我们将使用 mcp 库提供的 FastMCP 来快速构建一个服务器,它封装了加法计算、获取随机“秘密词”和查询天气三个工具。

创建一个名为 mcp_server.py 的新文件,我们将逐步实现它。

首先,导入必要的模块并初始化服务器:

import random
import requests
from mcp.server.fastmcp import FastMCP
import json

# 初始化一个FastMCP服务器实例,并给它起个名字
mcp = FastMCP("My First Tool Server")

FastMCP 是一个高级封装,它让创建MCP工具变得像写Python函数一样简单。

然后,开始定义我们的工具。每个工具都是一个用 @mcp.tool() 装饰器装饰的普通Python函数。函数的文档字符串(docstring)和类型注解至关重要,它们会被自动转换为MCP协议所需的工具描述,供AI模型理解。

@mcp.tool()
def add(a: int, b: int) -> int:
    """
    计算两个整数的和。

    参数:
        a (int): 第一个加数。
        b (int): 第二个加数。

    返回:
        int: a 与 b 的和。
    """
    # 在实际服务器中,这里可以加入日志
    print(f"[Server Log] 执行加法: {a} + {b}")
    return a + b

@mcp.tool()
def get_secret_word() -> str:
    """
    从一个预定义的列表中随机返回一个“秘密”单词。
    这个工具模拟了访问受保护或动态内容的过程。

    返回:
        str: 一个随机的单词,例如 'apple', 'banana', 或 'cherry'。
    """
    words = ["apple", "banana", "cherry", "dragonfruit", "elderberry"]
    chosen = random.choice(words)
    print(f"[Server Log] 随机选择了秘密单词: {chosen}")
    return chosen

前两个工具比较简单。第三个工具 get_current_weather 会复杂一些,因为它需要调用外部API。这里我们使用一个免费的公共服务 wttr.in 来获取天气信息。注意,我们增加了错误处理,使服务器更健壮。

@mcp.tool()
def get_current_weather(city: str, format: str = "3") -> str:
    """
    获取指定城市的当前天气情况。

    参数:
        city (str): 城市名称,例如 "Beijing", "Tokyo"。
        format (str): 输出格式。'3' 表示三天预报的简洁版。默认为 '3'。

    返回:
        str: 指定城市的天气信息文本。
    """
    print(f"[Server Log] 查询天气: 城市={city}, 格式={format}")
    endpoint = "https://wttr.in"

    try:
        # 设置User-Agent,因为wttr.in要求非命令行访问时最好设置
        headers = {'User-Agent': 'curl/7.68.0'} # 模拟curl请求
        response = requests.get(f"{endpoint}/{city}?format={format}", headers=headers, timeout=10)
        response.raise_for_status() # 如果HTTP请求返回错误状态码,则抛出异常
        return response.text
    except requests.exceptions.RequestException as e:
        # 将网络或API错误转化为对用户友好的信息
        error_msg = f"无法获取{city}的天气信息。错误: {e}"
        print(f"[Server Error] {error_msg}")
        return error_msg

最后,添加服务器启动代码:

if __name__ == "__main__":
    # 以SSE (Server-Sent Events) 传输方式运行服务器
    # 这将启动一个本地HTTP服务器,默认监听 http://0.0.0.0:8000
    mcp.run(transport="sse")

SSE是一种轻量级的、基于HTTP的服务器向客户端推送数据的技术,非常适合MCP这种需要服务器主动向客户端(Agent)通知工具列表变化的场景。

现在,在终端中运行你的服务器:

python mcp_server.py

如果一切正常,你会看到类似以下的输出,表明你的MCP服务器已经在 http://localhost:8000 上运行,并等待连接。

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

你的工具库已经上线了!下一步,就是创建一个智能体来使用它们。

4. 创建智能体客户端:连接、对话与工具调用

有了在后台运行的MCP服务器,我们现在需要构建客户端——也就是我们的AI智能体。它将连接到这个服务器,获取工具列表,并根据用户的指令智能地调用它们。

创建一个名为 agent_client.py 的新文件。我们将使用 openai-agents SDK来构建智能体。

首先,进行必要的导入和环境变量加载:

import asyncio
import os
from agents import Agent, OpenAIProvider, RunConfig, Runner
from agents.mcp import MCPServerSse  # 用于连接SSE类型的MCP服务器
from agents.model_settings import ModelSettings
from dotenv import load_dotenv

# 加载 .env 文件中的环境变量(主要是OPENAI_API_KEY)
load_dotenv()

接下来,我们编写一个异步函数 run_agent,它接受一个MCP服务器连接,并利用该连接创建一个智能体来执行任务。

async def run_agent(mcp_server):
    """
    使用给定的MCP服务器连接创建并运行一个智能体。

    参数:
        mcp_server: 一个已连接的MCPServer实例。
    """
    # 1. 创建智能体(Agent)
    agent = Agent(
        name="全能助手",
        instructions="你是一个乐于助人的助手,可以使用多种工具来回答问题。请根据用户的问题,判断是否需要使用工具,并选择最合适的工具。如果使用工具,请清晰地向用户展示结果。",
        mcp_servers=[mcp_server],  # 关键!将MCP服务器连接注入智能体
        model_settings=ModelSettings(
            tool_choice="auto",  # 让模型自动决定是否及如何使用工具
            temperature=0.1,     # 较低的温度使输出更确定、更专注
        )
    )

    # 2. 配置运行参数(RunConfig)
    run_config = RunConfig(
        model="gpt-4o",  # 指定使用的模型,gpt-4o在工具调用方面表现优异
        model_provider=OpenAIProvider(
            api_key=os.environ.get("OPENAI_API_KEY"),
            # 如果使用非OpenAI官方端点,可以在这里设置 base_url
            # base_url=os.environ.get("OPENAI_API_BASE"),
        ),
    )

    # 3. 定义一系列测试问题
    test_queries = [
        "请问331加上79等于多少?",
        "东京现在的天气怎么样?",
        "告诉我一个秘密单词。",
        "先查一下巴黎的天气,然后告诉我一个秘密单词。",
    ]

    # 4. 循环执行每个查询
    for query in test_queries:
        print(f"\n{'='*50}")
        print(f"用户提问: {query}")
        print(f"{'='*50}")

        try:
            # 使用Runner来运行智能体
            result = await Runner.run(
                starting_agent=agent,
                input=query,
                run_config=run_config,
            )
            # 打印智能体的最终输出
            print(f"助手回复:\n{result.final_output}")
            # 可以查看更详细的步骤(可选,用于调试)
            # for step in result.steps:
            #     print(f"步骤: {step.type} -> {step.output}")
        except Exception as e:
            print(f"执行过程中出现错误: {e}")

在这个函数中,有几个关键点:

  • mcp_servers=[mcp_server]:这是将工具能力赋予智能体的关键一行。智能体会自动从这个服务器发现工具。
  • tool_choice="auto":我们信任模型,让它根据问题和上下文自行决定是否调用工具、调用哪个工具。
  • 我们准备了一组测试问题,涵盖了数学计算、信息查询、组合任务等场景。

最后,编写主函数来启动整个流程:

async def main():
    """
    主函数:建立MCP服务器连接,并运行智能体。
    """
    # 使用异步上下文管理器连接MCP服务器
    # 确保你的 mcp_server.py 正在 localhost:8000 运行
    async with MCPServerSse(
        name="本地工具服务器",
        params={
            "url": "http://localhost:8000/sse",  # MCP SSE 端点的标准路径
        },
    ) as server:
        print("成功连接到MCP服务器,开始运行智能体...")
        await run_agent(server)

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

现在,确保你的 mcp_server.py 仍在运行。然后打开一个新的终端,激活同一个虚拟环境,运行客户端:

python agent_client.py

你将看到智能体开始工作,依次处理每个问题。对于加法问题,它会调用 add 工具;对于天气问题,它会调用 get_current_weather 工具(你可能看到它返回的ASCII艺术风格的天气信息);对于秘密单词,它会调用 get_secret_word。最有趣的是最后一个组合问题,智能体通常会规划两个步骤,依次调用两个工具,并将结果整合到一个连贯的回复中。

这个流程完美展示了MCP+Agent SDK的自动化威力:你无需在客户端代码中硬编码“如果是天气问题就调用A函数,如果是计算问题就调用B函数”,智能体会自己分析、决策和执行。

5. 进阶实战与最佳实践

掌握了基础流程后,我们可以探讨一些更深入的话题,让你的AI工具服务更强大、更可靠。

工具设计的艺术:编写对AI友好的工具

不是所有函数都适合直接暴露为MCP工具。好的工具设计能极大提升智能体的使用效果。

  • 清晰的命名与描述:工具名和描述应直白易懂。calculate_monthly_revenuedo_finance_stuff 好得多。在描述中说明工具的用途、适用场景和限制。
  • 严谨的类型注解:Python的类型提示(str, int, List[str] 等)会被转换为JSON Schema,帮助AI理解参数格式。使用 typing 模块中的 Optional, Literal 等可以表达更复杂的约束。
  • 提供示例(如有必要):在复杂的工具中,考虑在描述或通过MCP的“提示”功能提供调用示例。
  • 错误处理与友好反馈:工具函数内部应有完善的异常捕获,并返回对终端用户和AI都友好的错误信息,而不是堆栈跟踪。

连接多个MCP服务器:构建工具生态

一个智能体可以同时连接多个MCP服务器,从而形成一个庞大的工具网络。例如,你可以有:

  • 一个服务器专门提供数据查询工具(数据库、CRM)。
  • 一个服务器专门提供业务操作工具(发送通知、创建工单)。
  • 一个服务器专门提供第三方API工具(天气、地图、支付)。

在创建 Agent 时,只需将多个服务器实例放入 mcp_servers 列表即可:

agent = Agent(
    name="企业助手",
    instructions="...",
    mcp_servers=[finance_server, crm_server, notification_server],
    # ... 其他配置
)

智能体会自动从所有服务器中聚合工具列表,并根据任务上下文选择最合适的工具。

性能与监控考量

在生产环境中,你需要考虑更多:

  • 超时与重试:在 Runner.run 或工具调用层面设置合理的超时,并考虑实现重试逻辑,尤其是对于网络调用。
  • 日志与追踪:为MCP服务器和Agent客户端添加结构化日志(如使用 structloglogging 模块),记录工具调用、参数、结果和耗时,便于调试和监控。
  • 成本控制:工具调用可能会触发外部API费用。对于可能产生高成本或大量请求的工具(如图像生成、复杂计算),可以考虑在工具层面或Agent调用前加入配额检查、审批流程或成本估算提示。

调试技巧

当工具调用不按预期工作时,可以按以下步骤排查:

  1. 检查MCP服务器:首先确认服务器是否正常运行,并输出了正确的工具列表。你可以尝试用简单的HTTP客户端(如 curl)访问 http://localhost:8000/tools(如果SSE服务器暴露了此端点)或查看服务器启动日志。
  2. 检查客户端连接:确保客户端连接的URL和端口与服务器匹配。
  3. 查看详细日志:在开发时,可以临时提高日志级别,或像我们在示例代码中那样,在工具函数内添加 print 语句,观察调用是否触发。
  4. 简化测试:用一个最简单的工具(如我们的 add)和一句明确的指令(“计算1+1”)来测试端到端流程是否通畅。
  5. 审查工具描述:有时问题出在工具的描述或参数定义不够清晰,导致AI模型无法正确匹配。尝试简化或重写工具的描述文档。

构建基于MCP和Agent SDK的服务,最让我兴奋的一点是它的“声明式”编程体验。你只需要专注于定义“工具能做什么”(What),而把“何时以及如何做”(When and How)的决策权交给更擅长此道的AI模型。这种范式转变,让开发智能应用的效率提升了不止一个量级。从我自己的项目经验来看,初期花在工具定义和协议理解上的时间,会在后续功能扩展和维护阶段加倍地节省回来。

Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐