OpenAI Agents SDK ให้วิธีที่ทรงพลังในการสร้างแอปพลิเคชัน AI ที่มีความสามารถในการทำงานร่วมกับตัวแทน หนึ่งในฟีเจอร์หลักคือการรองรับ Model Context Protocol (MCP) ซึ่งมีมาตรฐานว่าด้วยวิธีที่แอปพลิเคชันจะให้บริบทและเครื่องมือแก่ Large Language Models (LLMs) บทความนี้จะนำคุณผ่านการเชื่อมต่อ OpenAI Agent SDK กับ MCP Servers โดยมีขั้นตอนและตัวอย่างโค้ดที่ละเอียด
MCP เป็นโปรโตคอลแบบเปิดที่มีมาตรฐานว่าแอปพลิเคชันจะให้บริบทอย่างไรแก่ LLMs ให้นึกถึง MCP เสมือน "พอร์ต USB-C" สำหรับแอปพลิเคชัน AI - ให้วิธีการมาตรฐานในการเชื่อมต่อโมเดล AI เข้ากับแหล่งข้อมูลและเครื่องมือต่างๆ เช่นเดียวกับ USB-C ที่เชื่อมโยงอุปกรณ์ของคุณกับอุปกรณ์ต่อพ่วงต่าง ๆ MCP จะเชื่อมโยงโมเดล AI เข้ากับเครื่องมือและแหล่งข้อมูลที่หลากหลาย
ข้อกำหนด MCP กำหนดเซิร์ฟเวอร์สองประเภทตามกลไกการขนส่งของพวกเขา:
OpenAI Agent SDK ให้คลาสที่สัมพันธ์กับทั้งสองประเภทเซิร์ฟเวอร์นี้:
MCPServerStdio สำหรับเซิร์ฟเวอร์ stdio ภายในท้องถิ่นMCPServerSse สำหรับเซิร์ฟเวอร์ SSE ระยะไกลก่อนที่คุณจะเริ่ม ตรวจสอบให้แน่ใจว่าคุณมี:
ติดตั้ง Python 3.10 หรือสูงกว่า
ติดตั้ง OpenAI Agent SDK:
pip install openai-agents
สำหรับการใช้เซิร์ฟเวอร์ stdio ภายในท้องถิ่น คุณอาจต้องใช้เครื่องมือเพิ่มเติม เช่น npx (สำหรับเซิร์ฟเวอร์ MCP ที่ใช้ JavaScript)
เริ่มต้นด้วยการเชื่อมต่อกับเซิร์ฟเวอร์ MCP แบบ stdio ภายในท้องถิ่น เราจะใช้เซิร์ฟเวอร์ MCP แบบไฟล์ระบบเป็นตัวอย่าง
หากคุณวางแผนที่จะใช้เซิร์ฟเวอร์ MCP แบบไฟล์ระบบ คุณจะต้องมี Node.js และ NPX:
# ติดตั้ง Node.js (ถ้ายังไม่ได้ติดตั้ง)
# สำหรับ Ubuntu/Debian
sudo apt update
sudo apt install nodejs npm
# สำหรับ macOS
brew install node
# ตรวจสอบการติดตั้ง
node --version
npx --version
สร้างโครงสร้างโปรเจคพื้นฐาน:
my_agent_project/
├── sample_data/
│ ├── file1.txt
│ └── file2.md
├── main.py
└── README.md
นี่คือตัวอย่างการเชื่อมต่อกับเซิร์ฟเวอร์ MCP แบบไฟล์ระบบโดยใช้ 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():
# รับพาธไดเรกทอรีสำหรับข้อมูลตัวอย่าง
current_dir = os.path.dirname(os.path.abspath(__file__))
sample_data_dir = os.path.join(current_dir, "sample_data")
# สร้างและกำหนดค่า MCP server
async with MCPServerStdio(
name="Filesystem Server",
params={
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", sample_data_dir],
},
) as mcp_server:
# สสร้างตัวแทนที่ใช้ MCP server
agent = Agent(
name="FileAssistant",
instructions="ใช้เครื่องมือไฟล์ระบบเพื่ออ่านและวิเคราะห์ไฟล์ในไดเรกทอรี sample_data.",
mcp_servers=[mcp_server]
)
# Run the agent with a user query
message = "รายการไฟล์ทั้งหมดในไดเรกทอรีและสรุปเนื้อหาของพวกเขา."
print(f"กำลังรันคำถาม: {message}\\\\n")
# สร้าง trace ID สำหรับการแก้ปัญหา
trace_id = gen_trace_id()
with trace(workflow_name="MCP Example Filesystem", trace_id=trace_id):
print(f"ดู trace: <https://platform.openai.com/traces/{trace_id}\\\\n>")
result = await Runner.run(starting_agent=agent, input=message)
print(result.final_output)
if __name__ == "__main__":
# ตรวจสอบว่า npx ได้ถูกติดตั้งแล้วหรือไม่
if not shutil.which("npx"):
raise RuntimeError("npx ไม่ได้ติดตั้ง กรุณาติดตั้งด้วย `npm install -g npx`.")
asyncio.run(run_agent_with_mcp())
ในตัวอย่างนี้:
MCPServerStdio ที่ทำงานเป็นซับโปรเซสสำหรับเซิร์ฟเวอร์ไฟล์ระบบAgent ผ่านพารามิเตอร์ mcp_serverslist_tools() บน MCP server โดยอัตโนมัติเพื่อให้ LLM ทราบเกี่ยวกับเครื่องมือที่มีอยู่call_tool() บนเซิร์ฟเวอร์ตอนนี้เรามาดูวิธีการเชื่อมต่อกับเซิร์ฟเวอร์ MCP ระยะไกลโดยใช้ SSE:
SSE (Server-Sent Events) MCP servers ทำงานจากระยะไกลและเปิดเผยฟังก์ชันการทำงานผ่าน HTTP endpoints ไม่เหมือนเซิร์ฟเวอร์ stdio พวกเขาไม่ทำงานเป็นซับโปรเซสของแอปพลิเคชันของคุณ
นี่คือตัวอย่างโค้ดที่เชื่อมต่อกับเซิร์ฟเวอร์ 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():
# สร้างและกำหนดค่า SSE MCP server connection
async with MCPServerSse(
name="บริการสภาพอากาศ",
params={
"url": "<https://example.com/mcp/sse>",
# พารามิเตอร์การตรวจสอบสิทธิ์ที่ไม่บังคับ
"headers": {
"Authorization": "Bearer your_api_key_here"
}
},
) as mcp_server:
# สร้างตัวแทนที่ใช้เซิร์ฟเวอร์ MCP ระยะไกล
agent = Agent(
name="WeatherAssistant",
instructions="ใช้เครื่องมือสภาพอากาศเพื่อตอบคำถามเกี่ยวกับสภาพอากาศปัจจุบัน.",
mcp_servers=[mcp_server],
# บังคับให้ตัวแทนใช้เครื่องมือเมื่อมี
model_settings=ModelSettings(tool_choice="required")
)
# รัน agent ด้วยคำถามจากผู้ใช้
message = "วันนี้สภาพอากาศในโตเกียวเป็นอย่างไร?"
print(f"กำลังรันคำถาม: {message}\\\\n")
trace_id = gen_trace_id()
with trace(workflow_name="Weather MCP Example", trace_id=trace_id):
print(f"ดู trace: <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())
เพื่อให้เข้าใจวิธีการทำงานของ MCP ได้ดีขึ้น เรามาทำการสร้างเซิร์ฟเวอร์ MCP ที่ง่าย ที่นี่คือตัวอย่างของ MCP SSE server ขนาดเล็ก:
import asyncio
import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from typing import Dict, Any, List, Optional
app = FastAPI()
# กำหนดเครื่องมือที่เซิร์ฟเวอร์ MCP ของเราจะให้บริการ
TOOLS = [
{
"type": "function",
"function": {
"name": "add",
"description": "บวกเลขสองตัวเข้าด้วยกัน",
"parameters": {
"type": "object",
"properties": {
"a": {"type": "number", "description": "หมายเลขแรก"},
"b": {"type": "number", "description": "หมายเลขที่สอง"}
},
"required": ["a", "b"]
}
}
},
{
"type": "function",
"function": {
"name": "get_weather",
"description": "รับข้อมูลสภาพอากาศสำหรับสถานที่",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "ชื่อเมือง"}
},
"required": ["location"]
}
}
}
]
# นำเสนอฟังก์ชันการทำงานของเครื่องมือจริง
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":
# ในการใช้งานจริง คุณจะเรียก API ตรวจอากาศจริง
weather_data = {
"Tokyo": {"condition": "มีแดด", "temperature": 25},
"New York": {"condition": "มีเมฆ", "temperature": 18},
"London": {"condition": "มีฝน", "temperature": 15}
}
location = parameters["location"]
if location in weather_data:
return {"weather": weather_data[location]}
return {"error": f"ไม่มีข้อมูลสภาพอากาศสำหรับ {location}"}
return {"error": f"ไม่รู้จักเครื่องมือ: {tool_name}"}
async def sse_event_generator(request: Request):
# อ่านเนื้อหาของคำขอ
body_bytes = await request.body()
body = json.loads(body_bytes)
# จัดการกับการดำเนินการ MCP ที่แตกต่างกัน
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)
เพื่อปรับปรุงประสิทธิภาพ คุณสามารถเก็บข้อมูลรายการเครื่องมือจาก MCP servers:
# สร้าง MCP server ด้วยการเก็บข้อมูลเครื่องมือ
async with MCPServerSse(
name="Weather Service",
params={"url": "<https://example.com/mcp/sse>"},
cache_tools_list=True # เปิดใช้งานการเก็บข้อมูล
) as mcp_server:
# ใช้เซิร์ฟเวอร์เหมือนเดิม...
เมื่อ cache_tools_list=True ถูกตั้งค่า SDK จะเรียก list_tools() บน MCP server เพียงครั้งเดียวและใช้ผลลัพธ์ในการรันตัวแทนต่อไป ซึ่งจะลดความล่าช้า โดยเฉพาะสำหรับเซิร์ฟเวอร์ระยะไกล
หากจำเป็นต้องยกเลิกการเก็บข้อมูล:
mcp_server.invalidate_tools_cache()
OpenAI Agents SDK มีความสามารถในการติดตามที่บันทึกการดำเนินงาน MCP โดยอัตโนมัติ:
คุณสามารถดูการติดตามเหล่านี้ได้ที่ https://platform.openai.com/traces/ เมื่ อคุณใช้ context manager trace ดังที่แสดงในตัวอย่างด้านบน
คุณสามารถเชื่อมต่อผู้แทนของคุณกับเซิร์ฟเวอร์ MCP หลายตัวในเวลาเดียวกัน ทำให้มันเข้าถึงเครื่องมือที่กว้างขึ้น:
async def run_with_multiple_servers():
async with MCPServerStdio(
name="Filesystem",
params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./data"]}
) as fs_server, MCPServerSse(
name="Weather API",
params={"url": "<https://example.com/weather/mcp/sse>"}
) as weather_server:
# สร้างผู้แทนด้วยเซิร์ฟเวอร์ MCP ทั้งสองตัว
agent = Agent(
name="MultiToolAssistant",
instructions="ใช้เครื่องมือทั้งหมดที่มีเพื่อช่วยผู้ใช้.",
mcp_servers=[fs_server, weather_server]
)
# รันผู้แทน
result = await Runner.run(
starting_agent=agent,
input="ตรวจสอบสภาพอากาศในโตเกียวก่อน จากนั้นอ่านเนื้อหาของไฟล์ report.txt."
)
print(result.final_output)
เมื่อทำงานกับ MCP servers คุณอาจพบปัญหาหลายอย่าง นี่คือปัญหาที่พบบ่อยและวิธีการจัดการ:
หากเซิร์ฟเวอร์ MCP ของคุณไม่ตอบสนอง:
try:
async with MCPServerSse(
name="Weather Service",
params={"url": "<https://example.com/mcp/sse>"}
) as mcp_server:
# ใช้เซิร์ฟเวอร์...
except Exception as e:
print(f"เชื่อมต่อเซิร์ฟเวอร์ MCP ล้มเหลว: {e}")
# ดำเนินการกลยุทธ์สำรอง
เมื่อการดำเนินการเครื่องมือล้มเหลว ให้จัดการอย่างสวยงาม:
try:
result = await Runner.run(starting_agent=agent, input=user_query)
print(result.final_output)
except Exception as e:
print(f"พบข้อผิดพลาดระหว่างการดำเนินการตัวแทน: {e}")
# คุณอาจต้องตรวจสอบบันทึกการติดตามสำหรับข้อมูลข้อผิดพลาดที่ละเอียด
การสนับสนุน MCP ของ OpenAI Agents SDK ช่วยให้คุณขยายตัวแทนของคุณด้วยเครื่องมือและความสามารถที่หลากหลาย ไม่ว่าคุณจะใช้เซิร์ฟเวอร์ stdio ในท้องถิ่นหรือ endpoints SSE ระยะไกล การรวมเข้าก็ทำได้ง่ายและทรงพลัง
โดยการเชื่อมต่อกับ MCP servers ตัวแทนของคุณสามารถเข้าถึงระบบไฟล์ API สภาพอากาศ ฐานข้อมูล และเครื่องมือหรือแหล่งข้อมูลภายนอกอื่น ๆ ที่เปิดเผยอินเทอร์เฟซ MCP ความยืดหยุ่นนี้ทำให้ OpenAI Agent SDK เป็นฐานที่ทรงพลังสำหรับการสร้างแอปพลิเคชัน AI ที่สลับซับซ้อน
อย่าลืมใช้ฟีเจอร์ต่างๆ เช่น การเก็บข้อมูลเครื่องมือเพื่อปรับปรุงประสิทธิภาพ และใช้ความสามารถในการติดตามที่สร้างขึ้นเพื่อติดตามและตรวจสอบการมีปฏิสัมพันธ์ของตัวแทนกับ MCP servers
ขอให้สนุกกับการเขียนโค้ดด้วย OpenAI Agents และ MCP!
