OpenAI Agents SDK는 에이전트 기능을 갖춘 AI 애플리케이션을 구축하는 강력한 방법을 제공합니다. 주요 기능 중 하나는 모델 컨텍스트 프로토콜(MCP)에 대한 지원으로, 이는 애플리케이션이 대규모 언어 모델(LLM)에 컨텍스트와 도구를 제공하는 방식을 표준화합니다. 이 글에서는 OpenAI Agent SDK를 MCP 서버에 연결하는 방법에 대해 상세한 단계와 샘플 코드를 안내합니다.
MCP는 애플리케이션이 LLM에 컨텍스트를 제공하는 방식을 표준화하는 오픈 프로토콜입니다. MCP를 AI 애플리케이션을 위한 "USB-C 포트"처럼 생각해 보세요 - 이 프로토콜은 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 서버용).
로컬 stdio MCP 서버에 연결하는 것부터 시작해 보겠습니다. 파일 시스템 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
이제 MCPServerStdio를 사용하여 파일 시스템 MCP 서버에 연결하는 전체 예제를 보여드리겠습니다:
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 서버를 생성하고 구성
async with MCPServerStdio(
name="파일 시스템 서버",
params={
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", sample_data_dir],
},
) as mcp_server:
# MCP 서버를 사용하는 에이전트 생성
agent = Agent(
name="FileAssistant",
instructions="샘플 데이터 디렉토리에 있는 파일을 읽고 분석하기 위해 파일 시스템 도구를 사용하십시오.",
mcp_servers=[mcp_server]
)
# 사용자 쿼리로 에이전트 실행
message = "디렉토리의 모든 파일을 나열하고 그 내용을 요약하세요."
print(f"쿼리 실행 중: {message}\\\\n")
# 디버깅을 위한 트레이스 ID 생성
trace_id = gen_trace_id()
with trace(workflow_name="MCP 파일 시스템 예제", trace_id=trace_id):
print(f"트레이스 보기: <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 인스턴스를 생성하여 파일 시스템 MCP 서버를 서브 프로세스로 실행합니다.mcp_servers 매개변수를 통해 이 서버를 Agent 생성자에 전달합니다.list_tools()를 호출하여 LLM이 사용 가능한 도구를 인식하게 됩니다.call_tool()을 호출합니다.이제 SSE를 사용하여 원격 MCP 서버에 연결하는 방법을 살펴보겠습니다:
SSE(서버 전송 이벤트) MCP 서버는 원격으로 실행되며 HTTP 엔드포인트를 통해 기능을 노출합니다. 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 서버 연결 생성 및 구성
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")
)
# 사용자 쿼리로 에이전트 실행
message = "오늘 도쿄의 날씨는 어떤가요?"
print(f"쿼리 실행 중: {message}\\\\n")
trace_id = gen_trace_id()
with trace(workflow_name="날씨 MCP 예제", trace_id=trace_id):
print(f"트레이스 보기: <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 서버의 예입니다:
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 서버에서 도구 목록을 캐시할 수 있습니다:
# 도구 캐싱으로 MCP 서버 생성
async with MCPServerSse(
name="날씨 서비스",
params={"url": "<https://example.com/mcp/sse>"},
cache_tools_list=True # 캐싱 활성화
) as mcp_server:
# 이전과 같이 서버 사용...
cache_tools_list=True로 설정하면 SDK가 MCP 서버에서 list_tools()를 한 번만 호출하고 후속 에이전트 실행 시 결과를 재사용합니다. 이는 특히 원격 서버의 경우 지연 시간을 줄여줍니다.
필요한 경우 캐시를 무효화하려면:
mcp_server.invalidate_tools_cache()
OpenAI Agents SDK는 MCP 작업을 자동으로 캡처하는 내장 추적 기능을 포함하고 있습니다:
위 예제에서 보듯이 trace 컨텍스트 관리자를 사용할 때 https://platform.openai.com/traces/에서 이러한 추적을 볼 수 있습니다.
에이전트를 여러 MCP 서버에 동시에 연결할 수 있어 더 넓은 범위의 도구에 접근할 수 있습니다:
async def run_with_multiple_servers():
async with MCPServerStdio(
name="파일 시스템",
params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./data"]}
) as fs_server, MCPServerSse(
name="날씨 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 서버 작업 시 다양한 문제에 직면할 수 있습니다. 다음은 일반적인 문제 및 처리 방법입니다:
MCP 서버가 응답하지 않는 경우:
try:
async with MCPServerSse(
name="날씨 서비스",
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}")
# 자세한 오류 정보를 위해 트레이스 로그 확인
OpenAI Agents SDK의 MCP 지원을 통해 에이전트를 다양한 도구와 기능으로 확장할 수 있습니다. 로컬 stdio 서버나 원격 SSE 엔드포인트를 사용하는 경우에도 통합은 간단하고 강력합니다.
MCP 서버에 연결하면 귀하의 에이전트가 파일 시스템, 날씨 API, 데이터베이스 및 MCP 인터페이스를 노출하는 사실상의 모든 외부 도구나 데이터 소스에 접근할 수 있습니다. 이러한 유연성은 OpenAI Agent SDK를 복잡한 AI 애플리케이션을 구축하는 데 강력한 기반이 되게 합니다.
실행 성능을 최적화하기 위해 도구 캐싱과 같은 기능을 활용하고, 에이전트가 MCP 서버와 상호작용하는 방법을 디버깅하고 모니터링하기 위해 내장 추적 기능을 사용하세요.
OpenAI Agents 및 MCP와 즐거운 코딩 되세요!
