SDK OpenAI Agents cung cấp một cách mạnh mẽ để xây dựng các ứng dụng AI với khả năng của tác nhân. Một trong những tính năng chính là hỗ trợ cho Giao thức Bối cảnh Mô hình (MCP), tiêu chuẩn hóa cách mà các ứng dụng cung cấp bối cảnh và công cụ cho các Mô hình Ngôn ngữ Lớn (LLMs). Bài viết này sẽ hướng dẫn bạn kết nối SDK Tác nhân OpenAI với các Máy chủ MCP, với các bước chi tiết và mã mẫu.
MCP là một giao thức mở tiêu chuẩn hóa cách mà các ứng dụng cung cấp bối cảnh cho LLMs. Hãy nghĩ về MCP như một "cổng USB-C" cho các ứng dụng AI - nó cung cấp một cách tiêu chuẩn để kết nối các mô hình AI với các nguồn dữ liệu và công cụ khác nhau. Cũng như USB-C kết nối các thiết bị của bạn với nhiều thiết bị ngoại vi khác nhau, MCP kết nối các mô hình AI với các công cụ và nguồn dữ liệu khác nhau.
Thông số kỹ thuật MCP định nghĩa hai loại máy chủ dựa trên cơ chế truyền tải của chúng:
SDK OpenAI Agent cung cấp các lớp tương ứng cho cả hai loại máy chủ này:
MCPServerStdio cho các máy chủ stdio cục bộMCPServerSse cho các máy chủ SSE từ xaTrước khi bắt đầu, hãy đảm bảo rằng bạn đã có:
Python 3.10 trở lên được cài đặt
SDK OpenAI Agent được cài đặt:
pip install openai-agents
Để sử dụng các máy chủ stdio cục bộ, bạn có thể cần các công cụ bổ sung như npx (cho các máy chủ MCP dựa trên JavaScript)
Hãy bắt đầu với việc kết nối đến một máy chủ MCP stdio cục bộ. Chúng ta sẽ sử dụng máy chủ MCP hệ thống tập tin làm ví dụ.
Nếu bạn dự định sử dụng máy chủ MCP hệ thống tập tin, bạn sẽ cần Node.js và NPX:
# Cài đặt Node.js (nếu chưa được cài đặt)
# Đối với Ubuntu/Debian
sudo apt update
sudo apt install nodejs npm
# Đối với macOS
brew install node
# Xác minh cài đặt
node --version
npx --version
Tạo một cấu trúc dự án cơ bản:
my_agent_project/
├── sample_data/
│ ├── file1.txt
│ └── file2.md
├── main.py
└── README.md
Dưới đây là một ví dụ hoàn chỉnh về việc kết nối đến máy chủ MCP hệ thống tập tin sử dụng MCPServerStdio:
import asyncio
import os
import shutil
from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServerStdio
async def run_agent_with_mcp():
# Lấy đường dẫn thư mục cho dữ liệu mẫu
current_dir = os.path.dirname(os.path.abspath(__file__))
sample_data_dir = os.path.join(current_dir, "sample_data")
# Tạo và cấu hình máy chủ MCP
async with MCPServerStdio(
name="Máy chủ Hệ thống tập tin",
params={
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", sample_data_dir],
},
) as mcp_server:
# Tạo một tác nhân sử dụng máy chủ MCP
agent = Agent(
name="Trợ lý Tập tin",
instructions="Sử dụng các công cụ hệ thống tập tin để đọc và phân tích các tệp trong thư mục sample_data.",
mcp_servers=[mcp_server]
)
# Chạy tác nhân với một câu hỏi người dùng
message = "Liệt kê tất cả các tệp trong thư mục và tóm tắt nội dung của chúng."
print(f"Đang chạy truy vấn: {message}\\\\n")
# Tạo ID theo dõi cho gỡ lỗi
trace_id = gen_trace_id()
with trace(workflow_name="Ví dụ MCP Hệ thống Tập tin", trace_id=trace_id):
print(f"Xem theo dõi: <https://platform.openai.com/traces/{trace_id}\\\\n>")
result = await Runner.run(starting_agent=agent, input=message)
print(result.final_output)
if __name__ == "__main__":
# Kiểm tra xem npx đã được cài đặt chưa
if not shutil.which("npx"):
raise RuntimeError("npx chưa được cài đặt. Vui lòng cài đặt nó bằng `npm install -g npx`.")
asyncio.run(run_agent_with_mcp())
Trong ví dụ này:
MCPServerStdio chạy máy chủ MCP hệ thống tập tin như một tiến trình con.Agent thông qua tham số mcp_servers.list_tools() trên máy chủ MCP để làm cho LLM nhận biết các công cụ có sẵn.call_tool() trên máy chủ.Giờ hãy xem cách kết nối với một máy chủ MCP từ xa sử dụng SSE:
Các máy chủ MCP SSE (Server-Sent Events) chạy từ xa và tiết lộ chức năng của chúng qua các điểm cuối HTTP. Khác với các máy chủ stdio, chúng không chạy như các tiến trình con của ứng dụng của bạn.
Dưới đây là một mã mẫu kết nối đến một máy chủ MCP SSE:
import asyncio
from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServerSse
from agents.model_settings import ModelSettings
async def run_agent_with_remote_mcp():
# Tạo và cấu hình kết nối máy chủ MCP SSE
async with MCPServerSse(
name="Dịch vụ Thời tiết",
params={
"url": "<https://example.com/mcp/sse>",
# Tham số xác thực tùy chọn
"headers": {
"Authorization": "Bearer your_api_key_here"
}
},
) as mcp_server:
# Tạo một tác nhân sử dụng máy chủ MCP từ xa
agent = Agent(
name="Trợ lý Thời tiết",
instructions="Sử dụng các công cụ thời tiết để trả lời các câu hỏi về điều kiện thời tiết hiện tại.",
mcp_servers=[mcp_server],
# Ép buộc tác nhân sử dụng công cụ khi có sẵn
model_settings=ModelSettings(tool_choice="required")
)
# Chạy tác nhân với một câu hỏi người dùng
message = "Thời tiết hôm nay ở Tokyo như thế nào?"
print(f"Đang chạy truy vấn: {message}\\\\n")
trace_id = gen_trace_id()
with trace(workflow_name="Ví dụ MCP Thời tiết", trace_id=trace_id):
print(f"Xem theo dõi: <https://platform.openai.com/traces/{trace_id}\\\\n>")
result = await Runner.run(starting_agent=agent, input=message)
print(result.final_output)
if __name__ == "__main__":
asyncio.run(run_agent_with_remote_mcp())
Để hiểu rõ hơn về cách MCP hoạt động, sẽ có ích khi triển khai một máy chủ MCP đơn giản. Dưới đây là một ví dụ về một máy chủ MCP SSE tối thiểu:
import asyncio
import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from typing import Dict, Any, List, Optional
app = FastAPI()
# Định nghĩa các công cụ mà máy chủ MCP của chúng tôi sẽ cung cấp
TOOLS = [
{
"type": "function",
"function": {
"name": "add",
"description": "Cộng hai số lại với nhau",
"parameters": {
"type": "object",
"properties": {
"a": {"type": "number", "description": "Số thứ nhất"},
"b": {"type": "number", "description": "Số thứ hai"}
},
"required": ["a", "b"]
}
}
},
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Lấy thời tiết cho một địa điểm",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "Tên thành phố"}
},
"required": ["location"]
}
}
}
]
# Triển khai chức năng công cụ thực tế
async def call_tool(tool_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
if tool_name == "add":
return {"result": parameters["a"] + parameters["b"]}
elif tool_name == "get_weather":
# Trong một triển khai thực tế, bạn sẽ gọi một API thời tiết thực
weather_data = {
"Tokyo": {"condition": "Nắng", "temperature": 25},
"New York": {"condition": "Đầy mây", "temperature": 18},
"London": {"condition": "Mưa", "temperature": 15}
}
location = parameters["location"]
if location in weather_data:
return {"weather": weather_data[location]}
return {"error": f"Dữ liệu thời tiết không có sẵn cho {location}"}
return {"error": f"Công cụ không xác định: {tool_name}"}
async def sse_event_generator(request: Request):
# Đọc nội dung yêu cầu
body_bytes = await request.body()
body = json.loads(body_bytes)
# Xử lý các thao tác MCP khác nhau
if body["action"] == "list_tools":
yield f"data: {json.dumps({'tools': TOOLS})}\\\\n\\\\n"
elif body["action"] == "call_tool":
tool_name = body["tool_name"]
parameters = body["parameters"]
result = await call_tool(tool_name, parameters)
yield f"data: {json.dumps({'result': result})}\\\\n\\\\n"
@app.post("/sse")
async def sse_endpoint(request: Request):
return StreamingResponse(
sse_event_generator(request),
media_type="text/event-stream"
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Để cải thiện hiệu suất, bạn có thể lưu cache danh sách công cụ từ các máy chủ MCP:
# Tạo một máy chủ MCP với lưu cache công cụ
async with MCPServerSse(
name="Dịch vụ Thời tiết",
params={"url": "<https://example.com/mcp/sse>"},
cache_tools_list=True # Bật lưu cache
) as mcp_server:
# Sử dụng máy chủ như trước...
Khi cache_tools_list=True được thiết lập, SDK sẽ chỉ gọi list_tools() trên máy chủ MCP một lần và tái sử dụng kết quả cho các lần chạy tác nhân sau. Điều này giảm độ trễ, đặc biệt là cho các máy chủ từ xa.
Để vô hiệu hóa cache nếu cần:
mcp_server.invalidate_tools_cache()
SDK OpenAI Agents bao gồm khả năng theo dõi tích hợp tự động ghi lại các thao tác MCP:
Bạn có thể xem những theo dõi này tại https://platform.openai.com/traces/ khi bạn sử dụng bộ quản lý ngữ cảnh trace như đã nêu trong các ví dụ ở trên.
Bạn có thể kết nối tác nhân của mình với nhiều máy chủ MCP cùng một lúc, giúp nó có quyền truy cập vào một loạt các công cụ rộng lớn hơn:
async def run_with_multiple_servers():
async with MCPServerStdio(
name="Hệ thống tập tin",
params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./data"]}
) as fs_server, MCPServerSse(
name="API Thời tiết",
params={"url": "<https://example.com/weather/mcp/sse>"}
) as weather_server:
# Tạo tác nhân với cả hai máy chủ MCP
agent = Agent(
name="Trợ lý Đa Công cụ",
instructions="Sử dụng tất cả các công cụ có sẵn để giúp người dùng.",
mcp_servers=[fs_server, weather_server]
)
# Chạy tác nhân
result = await Runner.run(
starting_agent=agent,
input="Đầu tiên kiểm tra thời tiết ở Tokyo, sau đó đọc nội dung của tệp report.txt."
)
print(result.final_output)
Khi làm việc với các máy chủ MCP, bạn có thể gặp phải nhiều vấn đề khác nhau. Dưới đây là một số lỗi thường gặp và cách xử lý chúng:
Nếu máy chủ MCP của bạn không phản hồi:
try:
async with MCPServerSse(
name="Dịch vụ Thời tiết",
params={"url": "<https://example.com/mcp/sse>"}
) as mcp_server:
# Sử dụng máy chủ...
except Exception as e:
print(f"Kết nối đến máy chủ MCP thất bại: {e}")
# Triển khai chiến lược dự phòng
Khi một lần thực thi công cụ thất bại, hãy xử lý nó một cách nhẹ nhàng:
try:
result = await Runner.run(starting_agent=agent, input=user_query)
print(result.final_output)
except Exception as e:
print(f"Lỗi trong quá trình thực thi tác nhân: {e}")
# Bạn có thể muốn kiểm tra nhật ký theo dõi để xem thông tin lỗi chi tiết
Hỗ trợ của SDK OpenAI Agents cho MCP cho phép bạn mở rộng các tác nhân của mình với nhiều công cụ và khả năng. Dù bạn đang sử dụng các máy chủ stdio cục bộ hay các điểm cuối SSE từ xa, việc tích hợp là dễ dàng và mạnh mẽ.
Bằng cách kết nối với các máy chủ MCP, các tác nhân của bạn có thể truy cập vào hệ thống tập tin, API thời tiết, cơ sở dữ liệu và hầu như bất kỳ công cụ hoặc nguồn dữ liệu bên ngoài nào có giao diện MCP. Sự linh hoạt này làm cho SDK OpenAI Agent trở thành nền tảng mạnh mẽ cho việc xây dựng các ứng dụng AI phức tạp.
Hãy nhớ tận dụng các tính năng như lưu cache công cụ để tối ưu hóa hiệu suất, và sử dụng các tính năng theo dõi tích hợp để gỡ lỗi và giám sát các tương tác của tác nhân với các máy chủ MCP.
Chúc bạn lập trình vui vẻ với OpenAI Agents và MCP!
