튜토리얼: DSPy에서 MCP 도구 사용하기
튜토리얼: DSPy에서 MCP 도구 사용하기 (Use MCP tools in DSPy)
MCP는 Model Context Protocol의 약자로, 애플리케이션이 LLM에 컨텍스트를 제공하는 방식을 표준화하는 공개 프로토콜이에요. 개발 오버헤드가 조금 있긴 하지만, MCP는 사용 중인 기술 스택과 무관하게 도구, 리소스, 프롬프트를 다른 개발자와 공유할 수 있는 귀중한 기회를 제공해요. 마찬가지로 다른 개발자가 만든 도구를 코드를 다시 작성하지 않고도 사용할 수 있습니다.
이 가이드에서는 DSPy에서 MCP 도구를 사용하는 방법을 안내할게요. 데모 목적으로 사용자가 항공편을 예약하고 기존 예약을 수정하거나 취소할 수 있게 도와주는 항공 서비스 에이전트를 구축할 거예요. 이는 커스텀 도구를 가진 MCP 서버에 의존하지만, 커뮤니티에서 만든 MCP 서버로 쉽게 일반화할 수 있어요.
??? "이 튜토리얼을 실행하는 방법" 이 튜토리얼은 Google Colab이나 Databricks notebook 같은 호스팅 IPython 노트북에서는 실행할 수 없어요. 코드를 실행하려면 가이드를 따라 로컬 기기에서 코드를 작성해야 합니다. 코드는 macOS에서 테스트되었으며 Linux 환경에서도 동일하게 동작할 거예요.
출처: 문서
본문
의존성 설치 (Install Dependencies)
시작하기 전에 필요한 의존성을 설치할게요:
pip install -U "dspy[mcp]"
MCP 서버 설정 (MCP Server Setup)
먼저 항공 에이전트를 위한 MCP 서버를 설정할게요. 서버에는 다음이 포함돼요:
- 데이터베이스 집합
- 사용자 데이터베이스: 사용자 정보를 저장해요.
- 항공편 데이터베이스: 항공편 정보를 저장해요.
- 티켓 데이터베이스: 고객 티켓을 저장해요.
- 도구 집합
- fetch_flight_info: 특정 날짜의 항공편 정보를 가져와요.
- fetch_itinerary: 예약된 일정(itinerary) 정보를 가져와요.
- book_itinerary: 사용자 대신 항공편을 예약해요.
- modify_itinerary: 항공편 변경이나 취소를 통해 일정을 수정해요.
- get_user_info: 사용자 정보를 가져와요.
- file_ticket: 사람의 도움이 필요할 때 백로그 티켓을 제기해요.
작업 디렉터리에 mcp_server.py 파일을 만들고 다음 내용을 붙여넣으세요:
import random
import string
from pydantic import BaseModel
try:
from mcp.server.fastmcp import FastMCP as MCPServer # SDK v1
except ImportError:
from mcp.server import MCPServer # SDK v2
# Create an MCP server
mcp = MCPServer("Airline Agent")
class Date(BaseModel):
# Somehow LLM is bad at specifying `datetime.datetime`
year: int
month: int
day: int
hour: int
class UserProfile(BaseModel):
user_id: str
name: str
email: str
class Flight(BaseModel):
flight_id: str
date_time: Date
origin: str
destination: str
duration: float
price: float
class Itinerary(BaseModel):
confirmation_number: str
user_profile: UserProfile
flight: Flight
class Ticket(BaseModel):
user_request: str
user_profile: UserProfile
user_database = {
"Adam": UserProfile(user_id="1", name="Adam", email="[email protected]"),
"Bob": UserProfile(user_id="2", name="Bob", email="[email protected]"),
"Chelsie": UserProfile(user_id="3", name="Chelsie", email="[email protected]"),
"David": UserProfile(user_id="4", name="David", email="[email protected]"),
}
flight_database = {
"DA123": Flight(
flight_id="DA123",
origin="SFO",
destination="JFK",
date_time=Date(year=2025, month=9, day=1, hour=1),
duration=3,
price=200,
),
"DA125": Flight(
flight_id="DA125",
origin="SFO",
destination="JFK",
date_time=Date(year=2025, month=9, day=1, hour=7),
duration=9,
price=500,
),
"DA456": Flight(
flight_id="DA456",
origin="SFO",
destination="SNA",
date_time=Date(year=2025, month=10, day=1, hour=1),
duration=2,
price=100,
),
"DA460": Flight(
flight_id="DA460",
origin="SFO",
destination="SNA",
date_time=Date(year=2025, month=10, day=1, hour=9),
duration=2,
price=120,
),
}
itinery_database = {}
ticket_database = {}
@mcp.tool()
def fetch_flight_info(date: Date, origin: str, destination: str):
"""Fetch flight information from origin to destination on the given date"""
flights = []
for flight_id, flight in flight_database.items():
if (
flight.date_time.year == date.year
and flight.date_time.month == date.month
and flight.date_time.day == date.day
and flight.origin == origin
and flight.destination == destination
):
flights.append(flight)
return flights
@mcp.tool()
def fetch_itinerary(confirmation_number: str):
"""Fetch a booked itinerary information from database"""
return itinery_database.get(confirmation_number)
@mcp.tool()
def pick_flight(flights: list[Flight]):
"""Pick up the best flight that matches users' request."""
sorted_flights = sorted(
flights,
key=lambda x: (
x.get("duration") if isinstance(x, dict) else x.duration,
x.get("price") if isinstance(x, dict) else x.price,
),
)
return sorted_flights[0]
def generate_id(length=8):
chars = string.ascii_lowercase + string.digits
return "".join(random.choices(chars, k=length))
@mcp.tool()
def book_itinerary(flight: Flight, user_profile: UserProfile):
"""Book a flight on behalf of the user."""
confirmation_number = generate_id()
while confirmation_number in itinery_database:
confirmation_number = generate_id()
itinery_database[confirmation_number] = Itinerary(
confirmation_number=confirmation_number,
user_profile=user_profile,
flight=flight,
)
return confirmation_number, itinery_database[confirmation_number]
@mcp.tool()
def cancel_itinerary(confirmation_number: str, user_profile: UserProfile):
"""Cancel an itinerary on behalf of the user."""
if confirmation_number in itinery_database:
del itinery_database[confirmation_number]
return
raise ValueError("Cannot find the itinerary, please check your confirmation number.")
@mcp.tool()
def get_user_info(name: str):
"""Fetch the user profile from database with given name."""
return user_database.get(name)
@mcp.tool()
def file_ticket(user_request: str, user_profile: UserProfile):
"""File a customer support ticket if this is something the agent cannot handle."""
ticket_id = generate_id(length=6)
ticket_database[ticket_id] = Ticket(
user_request=user_request,
user_profile=user_profile,
)
return ticket_id
if __name__ == "__main__":
mcp.run()
서버를 시작하기 전에 코드를 잠시 살펴볼게요.
먼저 설치된 SDK 버전이 제공하는 클래스 이름을 사용해 서버 인스턴스를 만들어요:
mcp = MCPServer("Airline Agent")
그런 다음 데이터 구조를 정의하는데, 실제 애플리케이션에서는 이것이 데이터베이스 스키마가 될 거예요. 예를 들어:
class Flight(BaseModel):
flight_id: str
date_time: Date
origin: str
destination: str
duration: float
price: float
이어서 데이터베이스 인스턴스를 초기화해요. 실제 애플리케이션에서는 이것들이 실제 데이터베이스에 대한 커넥터가 되겠지만, 단순함을 위해 여기서는 사전(dictionary)만 사용해요:
user_database = {
"Adam": UserProfile(user_id="1", name="Adam", email="[email protected]"),
"Bob": UserProfile(user_id="2", name="Bob", email="[email protected]"),
"Chelsie": UserProfile(user_id="3", name="Chelsie", email="[email protected]"),
"David": UserProfile(user_id="4", name="David", email="[email protected]"),
}
다음 단계는 도구를 정의하고 @mcp.tool()로 표시해서 MCP 클라이언트가 이들을 MCP 도구로 발견할 수 있게 하는 거예요:
@mcp.tool()
def fetch_flight_info(date: Date, origin: str, destination: str):
"""Fetch flight information from origin to destination on the given date"""
flights = []
for flight_id, flight in flight_database.items():
if (
flight.date_time.year == date.year
and flight.date_time.month == date.month
and flight.date_time.day == date.day
and flight.origin == origin
and flight.destination == destination
):
flights.append(flight)
return flights
마지막 단계는 서버를 띄우는 거예요:
if __name__ == "__main__":
mcp.run()
이제 서버 작성이 끝났어요. 별도로 실행할 필요는 없습니다. 아래의 MCP 클라이언트가 서버를 하위 프로세스로 시작하거든요.
MCP 서버의 도구를 활용하는 DSPy 프로그램 작성 (Write a DSPy Program That Utilizes Tools in MCP Server)
서버가 준비됐으니, 서버의 MCP 도구를 활용해 사용자를 돕는 실제 항공 서비스 에이전트를 구축해볼게요. 작업 디렉터리에 dspy_mcp_agent.py 파일을 만들고 가이드를 따라 코드를 추가해 보세요.
MCP 서버에서 도구 수집 (Gather Tools from MCP Servers)
먼저 MCP 서버에서 사용 가능한 모든 도구를 수집하고 DSPy에서 사용할 수 있게 만들어야 해요. DSPy는 표준 도구 인터페이스로 dspy.Tool API를 제공합니다. 모든 MCP 도구를 dspy.Tool로 변환해볼게요.
MCP SDK v2의 고수준 Client를 사용해 서버를 시작하고 사용 가능한 도구를 가져온 뒤, 정적 메서드 from_mcp_tool을 사용해 dspy.Tool로 변환해요. DSPy가 변환된 도구를 사용하는 동안 클라이언트 컨텍스트를 열어 둡니다:
from mcp import Client, StdioServerParameters
from mcp.client.stdio import stdio_client
import dspy
# Create server parameters for stdio connection
server_params = StdioServerParameters(
command="python", # Executable
args=["path_to_your_working_directory/mcp_server.py"],
env=None,
)
async def run():
async with Client(stdio_client(server_params)) as client:
# List available tools
tools = await client.list_tools()
# Convert MCP tools to DSPy tools
dspy_tools = []
for tool in tools.tools:
dspy_tools.append(dspy.Tool.from_mcp_tool(client, tool))
print(len(dspy_tools))
print(dspy_tools[0].args)
if __name__ == "__main__":
import asyncio
asyncio.run(run())
위 코드로 사용 가능한 모든 MCP 도구를 성공적으로 수집하고 DSPy 도구로 변환했어요.
고객 요청을 처리하는 DSPy 에이전트 구축 (Build a DSPy Agent to Handle Customer Requests)
이제 dspy.ReAct를 사용해 고객 요청을 처리하는 에이전트를 구축할 거예요. ReAct는 "reasoning and acting"(추론과 행동)의 약자로, LLM에게 도구를 호출할지 프로세스를 마무리할지 결정하도록 요청해요. 도구가 필요하다면 어떤 도구를 호출할지와 적절한 인자를 제공할 책임은 LLM이 집니다.
평소처럼 에이전트의 입력과 출력을 정의하는 dspy.Signature를 만들어야 해요:
import dspy
class DSPyAirlineCustomerService(dspy.Signature):
"""You are an airline customer service agent. You are given a list of tools to handle user requests. You should decide the right tool to use in order to fulfill users' requests."""
user_request: str = dspy.InputField()
process_result: str = dspy.OutputField(
desc=(
"Message that summarizes the process result, and the information users need, "
"e.g., the confirmation_number if it's a flight booking request."
)
)
그리고 에이전트의 LM을 선택해요:
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
그런 다음 도구와 시그니처를 dspy.ReAct API에 전달해 ReAct 에이전트를 만듭니다. 이제 완전한 코드 스크립트를 정리해볼게요:
from mcp import Client, StdioServerParameters
from mcp.client.stdio import stdio_client
import dspy
# Create server parameters for stdio connection
server_params = StdioServerParameters(
command="python", # Executable
args=["path_to_your_working_directory/mcp_server.py"], # Server script
env=None, # Optional environment variables
)
class DSPyAirlineCustomerService(dspy.Signature):
"""You are an airline customer service agent. You are given a list of tools to handle user requests.
You should decide the right tool to use in order to fulfill users' requests."""
user_request: str = dspy.InputField()
process_result: str = dspy.OutputField(
desc=(
"Message that summarizes the process result, and the information users need, "
"e.g., the confirmation_number if it's a flight booking request."
)
)
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
async def run(user_request):
async with Client(stdio_client(server_params)) as client:
# List available tools
tools = await client.list_tools()
# Convert MCP tools to DSPy tools
dspy_tools = []
for tool in tools.tools:
dspy_tools.append(dspy.Tool.from_mcp_tool(client, tool))
# Create the agent
react = dspy.ReAct(DSPyAirlineCustomerService, tools=dspy_tools)
result = await react.acall(user_request=user_request)
print(result)
if __name__ == "__main__":
import asyncio
asyncio.run(run("please help me book a flight from SFO to JFK on 09/01/2025, my name is Adam"))
MCP 도구는 기본적으로 async이기 때문에 react.acall을 호출해야 한다는 점에 유의하세요. 스크립트를 실행해볼게요:
python path_to_your_working_directory/dspy_mcp_agent.py
다음과 비슷한 출력을 볼 수 있어요:
Prediction(
trajectory={'thought_0': 'I need to fetch flight information for Adam from SFO to JFK on 09/01/2025 to find available flights for booking.', 'tool_name_0': 'fetch_flight_info', 'tool_args_0': {'date': {'year': 2025, 'month': 9, 'day': 1, 'hour': 0}, 'origin': 'SFO', 'destination': 'JFK'}, 'observation_0': ['{"flight_id": "DA123", "date_time": {"year": 2025, "month": 9, "day": 1, "hour": 1}, "origin": "SFO", "destination": "JFK", "duration": 3.0, "price": 200.0}', '{"flight_id": "DA125", "date_time": {"year": 2025, "month": 9, "day": 1, "hour": 7}, "origin": "SFO", "destination": "JFK", "duration": 9.0, "price": 500.0}'], ..., 'tool_name_4': 'finish', 'tool_args_4': {}, 'observation_4': 'Completed.'},
reasoning="I successfully booked a flight for Adam from SFO to JFK on 09/01/2025. I found two available flights, selected the more economical option (flight DA123 at 1 AM for $200), retrieved Adam's user profile, and completed the booking process. The confirmation number for the flight is 8h7clk3q.",
process_result='Your flight from SFO to JFK on 09/01/2025 has been successfully booked. Your confirmation number is 8h7clk3q.'
)
trajectory 필드는 전체 사고 및 행동 과정을 담고 있어요. 내부에서 무슨 일이 일어나는지 궁금하다면 Observability Guide를 참고해 MLflow를 설정해 보세요. dspy.ReAct 내부에서 일어나는 모든 단계를 시각화해 줍니다!
결론 (Conclusion)
이 가이드에서는 커스텀 MCP 서버와 dspy.ReAct 모듈을 활용하는 항공 서비스 에이전트를 구축했어요. MCP 지원 맥락에서 DSPy는 MCP 도구와 상호작용하는 간단한 인터페이스를 제공해, 필요한 모든 기능을 유연하게 구현할 수 있게 해줍니다.