Client
Client
이 문서에서는 Pydantic AI를 MCP 클라이언트로 사용해 MCP 서버에 연결하고 그 도구를 에이전트 실행의 일부로 사용하는 방법을 알려드려요. MCPToolset toolset은 FastMCP Client를 감싸며 로컬(stdio)과 원격(Streamable HTTP, SSE) MCP 서버 모두에서 동작해요.
출처: 문서
본문
Pydantic AI는 MCP 클라이언트로 동작할 수 있으며, MCP 서버에 연결해 그 도구를 에이전트 실행의 일부로 사용해요. MCPToolset toolset은 FastMCP Client를 감싸며 로컬(stdio)과 원격(Streamable HTTP, SSE) MCP 서버 모두와 함께 동작해요.
권장: MCP capability
대부분의 사용 사례에서는 MCP capability를 사용하세요. URL(또는 local=로 어떤 MCPToolset 입력)을 받고, 단일 native=True 플래그로 모델 프로바이더의 네이티브 MCP 지원에 옵트인할 수 있게 해요. 클라이언트 수명 주기를 직접 관리하거나, 같은 MCP 서버를 여러 에이전트에 붙이거나, capability가 노출하지 않는 고급 전송·클라이언트 구성을 전달해야 할 때는 MCPToolset을 직접 사용하세요.
Install
pydantic-ai를 설치하거나, mcp 옵션 그룹과 함께 pydantic-ai-slim을 설치해야 해요:
Terminal
pip install "pydantic-ai-slim[mcp]"
Terminal
uv add "pydantic-ai-slim[mcp]"
FastMCP 4
위 명령은 MCPToolset이 FastMCP 3과 함께 지원하는 FastMCP 4를 설치해요. 그것의 현대 프로토콜 모드는 서버 주도 sampling이나 elicitation을 지원하지 않으며 log_level을 적용할 수 없어요. MCPToolset은 이러한 옵션이 구성되면 경고해요. log_handler에서 로그를 필터링하세요. FastMCP 4의 레거시 프로토콜 모드에서는 이러한 옵션에 대해 FastMCP 3 동작을 유지해요.
FastMCP 4는 또한 레거시 HTTPX가 아닌 httpx2 위에 구축돼요. MCPToolset은 전달한 auth와 http_client 객체를 검사 없이 FastMCP에 그대로 넘겨주므로, httpx2로 만들거나 — FastMCP 3을 고정했다면 레거시 httpx로 만드세요.
Usage
MCPToolset은 첫 번째 위치 인자로 다음 중 어떤 것이든 받아들여요:
- URL 문자열(Streamable HTTP, 또는 경로가
/sse로 끝나면 SSE) - 로컬 Python 또는 Node.js 스크립트 경로(stdio로 실행)
StdioTransport,StreamableHttpTransport,SSETransport같은 FastMCP transport- 사전 빌드된
fastmcp.Client(OAuth나 tool transformation 같은 고급 FastMCP 특정 구성용) - in-process FastMCP 서버(테스트 또는 단일 프로세스 배포용 — 네트워크 왕복 없음)
각 MCPToolset 인스턴스는 toolset이며 toolsets 인자를 통해 Agent에 등록할 수 있어요.
async with agent을 사용해 등록된 모든 MCP toolset(그리고 stdio 서버의 경우 하위 프로세스 시작·중지)에 대한 연결을 에이전트 실행에서 사용될 컨텍스트 주변에서 열고 닫을 수 있어요. 특정 toolset의 수명 주기를 직접 관리하려면 async with toolset을 사용할 수도 있어요(예: 여러 에이전트 간 공유). 이 컨텍스트 관리자 중 하나를 명시적으로 입력하지 않으면, toolset이 필요할 때 자동으로 열리고 닫혀요.
공유된 MCPToolset 인스턴스는 단일 정체성으로 서버에 연결한다는 점을 주의하세요. 사용자가 MCP 서버에 대한 자체 자격 증명을 가진다면 사용자별 인증을 참고하세요.
Streamable HTTP
Streamable HTTP transport는 원격 MCP 서버에 연결하는 권장 방식이에요.
Note
Streamable HTTP MCPToolset은 에이전트를 실행하기 전에 MCP 서버가 실행 중이고 HTTP 연결을 받아들이고 있어야 해요. 서버 실행은 Pydantic AI가 관리하지 않아요.
toolset을 만들기 전에 Streamable HTTP transport를 지원하는 서버를 실행해야 해요.
streamable_http_server.py
from mcp.server.fastmcp import FastMCP
app = FastMCP()
@app.tool()
def add(a: int, b: int) -> int:
return a + b
if __name__ == '__main__':
app.run(transport='streamable-http')
그런 다음 toolset을 만들 수 있어요:
mcp_streamable_http_client.py
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
toolset = MCPToolset('http://localhost:8000/mcp') # (1)
agent = Agent('openai:gpt-5.2', toolsets=[toolset]) # (2)
async def main():
result = await agent.run('What is 7 plus 5?')
print(result.output)
#> The answer is 12.
연결에 사용할 URL로 MCP toolset을 정의하세요.
MCP toolset을 붙인 에이전트를 만드세요.
(이 예시를 실행하려면 asyncio를 import하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
여기서 무슨 일이 일어나나요?
- 모델이 "What is 7 plus 5?" 프롬프트를 받아요
- 모델이 "아,
add도구가 있네, 이 질문에 답하는 좋은 방법이겠다"라고 결정해요 - 모델이 도구 호출을 반환해요
- Pydantic AI가 Streamable HTTP transport를 사용해 도구 호출을 MCP 서버로 보내요
add도구 실행의 반환 값(12)으로 모델이 다시 호출돼요- 모델이 최종 답을 반환해요
이를 명확히 시각화하고 도구 호출까지 볼 수 있게 하려면, 예시를 logfire로 계측하는 세 줄의 코드를 추가하세요:
mcp_streamable_http_client_logfire.py
import logfire
logfire.configure()
logfire.instrument_pydantic_ai()
SSE
HTTP + Server-Sent Events transport도 지원돼요. /sse로 끝나는 URL은 SSE로 자동 감지되고, 다른 경로에는 명시적 SSETransport를 전달하세요.
Note
MCP의 SSE transport는 deprecate됐어요. 새 배포에는 Streamable HTTP를 선호하세요.
mcp_sse_client.py
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
toolset = MCPToolset('http://localhost:3001/sse')
agent = Agent('openai:gpt-5.2', toolsets=[toolset])
Stdio
MCP는 또한 stdio transport를 제공하며, 여기서 서버는 하위 프로세스로 실행되고 stdin과 stdout으로 클라이언트와 통신해요. Python 또는 Node.js 스크립트의 경로를 전달하거나, 명령, 인자, 환경을 완전히 제어하려면 StdioTransport를 만드세요.
mcp_stdio_client.py
from fastmcp.client.transports import StdioTransport
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
toolset = MCPToolset(StdioTransport(command='python', args=['mcp_server.py']))
agent = Agent('openai:gpt-5.2', toolsets=[toolset])
In-process FastMCP server
에이전트와 같은 Python 프로세스에 이미 FastMCP 서버가 있다면, 그것을 MCPToolset에 직접 넘겨 네트워크 왕복을 절약할 수 있어요:
mcp_in_process_server.py
from fastmcp import FastMCP
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
fastmcp_server = FastMCP('my_server')
@fastmcp_server.tool()
async def add(a: int, b: int) -> int:
return a + b
toolset = MCPToolset(fastmcp_server)
agent = Agent('openai:gpt-5.2', toolsets=[toolset])
async def main():
result = await agent.run('What is 7 plus 5?')
print(result.output)
#> The answer is 12.
(이 예시를 실행하려면 asyncio를 import하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
Loading MCP toolsets from configuration
MCPToolset 인스턴스를 개별적으로 구성하는 대신, load_mcp_toolsets()을 사용해 JSON 구성 파일에서 여러 toolset을 로드할 수 있어요.
이는 여러 MCP 서버를 관리해야 하거나, 코드를 수정하지 않고 서버를 외부에서 구성하고 싶을 때 특히 유용해요.
Configuration format
구성 파일은 서버 정의를 담은 mcpServers 객체가 있는 JSON 파일이어야 해요. 각 서버는 고유한 키로 식별되며 해당 서버의 구성을 담아요:
mcp_config.json
{
"mcpServers": {
"python-runner": {
"command": "uv",
"args": ["run", "mcp-run-python", "stdio"]
},
"weather": {
"command": "python",
"args": ["mcp_server.py"]
},
"weather-api": {
"url": "http://localhost:3001/sse"
},
"calculator": {
"url": "http://localhost:8000/mcp"
}
}
}
각 항목은 stdio 서버용 command, args, env, cwd 또는 HTTP 서버용 url과 headers를 지원해요. 구성은 로드될 때 검증되므로, 잘못된 타입의 필드는 나중에 연결 시점에 실패하는 대신 즉시 보고돼요. 알 수 없는 키는 무시되므로, 다른 MCP 클라이언트와 공유된 파일도 여전히 로드돼요 — 하지만 무시될 뿐 절대 받아들여지지 않아요. 특히 disabled는 서버를 건너뛰지 않고, type은 transport를 선택하지 않아요. 그것은 아래 설명대로 URL에서 추론돼요.
Note
MCP 서버는 /sse 접미사 때문에만 SSE 서버로 추론돼요. url 필드가 있는 다른 서버는 Streamable HTTP 서버로 취급돼요. SSE transport가 deprecate됐다는 점을 고려해 이 결정을 내렸어요.
Environment variables
구성 파일은 ${VAR} 및 ${VAR:-default} 구문으로 환경 변수 확장을 지원해요, Claude Code처럼. 이는 API 키나 호스트 이름 같은 민감한 정보를 구성 파일 밖에 두는 데 유용해요:
mcp_config_with_env.json
{
"mcpServers": {
"python-runner": {
"command": "${PYTHON_CMD:-python3}",
"args": ["run", "${MCP_MODULE}", "stdio"],
"env": {
"API_KEY": "${MY_API_KEY}"
}
},
"weather-api": {
"url": "https://${SERVER_HOST:-localhost}:${SERVER_PORT:-8080}/sse"
}
}
}
load_mcp_toolsets()으로 이 구성을 로드할 때:
${VAR}참조는 해당 환경 변수 값으로 대체돼요.${VAR:-default}참조는 설정된 경우 환경 변수 값을, 그렇지 않으면 기본값을 사용해요.
Caution
${VAR} 구문을 사용하는 참조 환경 변수가 정의되지 않으면 ValueError가 발생해요. 폴백 값을 제공하려면 ${VAR:-default} 구문을 사용하세요.
구성 파일을 신뢰된 입력으로 취급하세요
구성 파일은 하위 프로세스로 실행할 실행 파일과 인자를 지정하므로, 이 파일을 쓸 수 있는 사람은 누구나 임의의 명령을 실행할 수 있어요. ${VAR} 참조는 allowlist 없이 전체 프로세스 환경에 대해 확장되므로, 구성 파일이 어떤 환경 변수도 읽을 수 있어요. 통제하는 구성 파일만 로드하세요. 신뢰할 수 없는 소스에서 절대 로드하지 마세요.
Usage
mcp_config_loader.py
from pydantic_ai import Agent
from pydantic_ai.mcp import load_mcp_toolsets
# Load all toolsets from the configuration file
toolsets = load_mcp_toolsets('mcp_config.json')
# Create an agent with all loaded toolsets
agent = Agent('openai:gpt-5.2', toolsets=toolsets)
async def main():
result = await agent.run('What is 7 plus 5?')
print(result.output)
Tool call customization
MCPToolset은 process_tool_call 콜백을 받아들이며, 이를 통해 도구 호출 요청과 그 응답을 맞춤 설정할 수 있어요. 흔한 사용 사례는 서버 측 핸들러가 읽어야 할 메타데이터를 주입하는 것이에요:
mcp_process_tool_call.py
from typing import Any
from fastmcp.client.transports import StdioTransport
from pydantic_ai import Agent, RunContext
from pydantic_ai.mcp import CallToolFunc, MCPToolset, ToolResult
from pydantic_ai.models.test import TestModel
async def process_tool_call(
ctx: RunContext[int],
call_tool: CallToolFunc,
name: str,
tool_args: dict[str, Any],
) -> ToolResult:
"""A tool call processor that passes along the deps."""
return await call_tool(name, tool_args, {'deps': ctx.deps})
toolset = MCPToolset(
StdioTransport(command='python', args=['mcp_server.py']),
process_tool_call=process_tool_call,
)
agent = Agent(
model=TestModel(call_tools=['echo_deps']),
deps_type=int,
toolsets=[toolset],
)
async def main():
result = await agent.run('Echo with deps set to 42', deps=42)
print(result.output)
#> {"echo_deps":{"echo":"This is an echo message","deps":42}}
서버가 주입된 메타데이터를 읽는 방법은 MCP 서버 SDK 특정이에요. 예를 들어 MCP Python SDK에서는 도구 핸들러의 ctx: Context 인자를 통해 접근할 수 있어요:
mcp_server.py
from typing import Any
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
mcp = FastMCP('Pydantic AI MCP Server')
@mcp.tool()
async def echo_deps(ctx: Context[ServerSession, None]) -> dict[str, Any]:
"""Echo the run context.
Args:
ctx: Context object containing request and session information.
Returns:
Dictionary with an echo message and the deps.
"""
await ctx.info('This is an info message')
deps: Any = getattr(ctx.request_context.meta, 'deps')
return {'echo': 'This is an echo message', 'deps': deps}
if __name__ == '__main__':
mcp.run()
Tool errors
MCP 서버가 도구 오류를 보고하면, MCPToolset은 그 오류가 모델에 재시도를 요청해야 하는지, 실패한 도구 결과로 나타나야 하는지, 예외로 빠져나가야 하는지 선택하게 해요:
tool_error_behavior
Behavior
'retry'
기본값. ModelRetry를 발생시켜 서버 오류를 재시도 프롬프트로 모델에 보내요. 모델이 호출을 수정할 수 있을 때 사용하세요.
'failed'
ToolFailed를 발생시키며 outcome='failed'인 도구 결과로 기록돼요. 도구 호출이 완료됐지만 실패했고, 모델이 다음에 무엇을 할지 결정해야 할 때 사용하세요.
'error'
기본 MCP 도구 예외를 전파하고 에이전트 실행을 실패시켜요. 모델 루프 밖에서 애플리케이션 코드가 처리하기를 원하는 오류에 사용하세요.
구조화된 오류 콘텐츠는 'retry'와 'failed' 둘 다에서 모델이 보는 메시지에 JSON으로 직렬화되므로, 재시도 가능성 힌트와 기타 기계 판독 가능한 세부사항이 모델에 계속 사용 가능해요. 프로토콜과 transport 오류는 완료된 실패 도구 호출로 보고되지 않아요.
이것은 로컬 도구 코드에서 도구 재시도 대 실패한 도구 결과 구분의 MCP 동등물이에요.
Tool prefixes to avoid naming conflicts
같은 이름의 도구를 제공할 수 있는 여러 MCP 서버에 연결할 때, 각 MCPToolset을 .prefixed(...)로 감싸 도구 이름 앞에 접두사를 붙이세요:
mcp_tool_prefix.py
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
weather = MCPToolset('http://localhost:3001/sse').prefixed('weather') # `weather_*`
calculator = MCPToolset('http://localhost:3002/sse').prefixed('calc') # `calc_*`
# Both servers may expose a `get_data` tool, but they're disambiguated as
# `weather_get_data` and `calc_get_data`.
agent = Agent('openai:gpt-5.2', toolsets=[weather, calculator])
Server instructions
MCP 서버는 초기화 중에 서버의 도구와 가장 잘 상호작용하는 방법에 대한 컨텍스트를 주는 지침을 제공할 수 있어요. 이는 연결이 확립된 후 MCPToolset.instructions로 접근할 수 있으며, include_instructions=True로 설정해 에이전트의 지침에 자동으로 주입할 수 있어요:
mcp_server_include_instructions.py
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
toolset = MCPToolset('http://localhost:8000/mcp', include_instructions=True)
agent = Agent('openai:gpt-5.2', toolsets=[toolset])
Tool metadata
MCP 도구는 도구의 특성에 대한 추가 정보를 주는 메타데이터를 포함할 수 있으며, 이는 도구 필터링에 유용해요. meta와 annotations 필드는 필터 함수에 전달되는 ToolDefinition 객체의 metadata dict에서 찾을 수 있고, 도구의 출력 스키마(있다면)는 return_schema 필드로 사용 가능해요.
MCPToolset은 또한 toolset이 task-augmented execution을 도구에 사용할지 나타내는 task: bool 플래그를 노출해요. task 지원이 선택적인 도구의 경우, 이는 prefer_tasks 설정을 반영해요.
Background tasks
MCPToolset은 MCP task-augmented execution(SEP-1686)을 지원해요. FastMCP 3 서버를 포함해 SEP-1686을 사용하는 서버는 execution.taskSupport로 도구별 task 지원을 선언할 수 있고, MCPToolset은 그에 따라 호출을 라우팅해요:
execution.taskSupport
Behavior
"required"
항상 task=True로 호출. 서버가 task를 만들고 클라이언트가 tasks/result로 최종 결과를 기다려요.
"optional"
기본적으로 task=True로 호출. 대신 정상 호출하려면 prefer_tasks=False를 설정하세요.
"forbidden" 또는 없음
정상 호출.
FastMCP 4는 서버가 task 생성을 지시하는 더 새로운 MCP Tasks extension(SEP-2663)을 사용해요. 따라서 위의 task 메타데이터와 prefer_tasks 클라이언트 선호는 FastMCP 4가 아니라 FastMCP 3에 적용돼요. 일반 호출은 추가 설치 없이 task 전용 도구를 완료로 몰고 가며, use_task=True로 tasks extension을 명시적으로 선택하려면 fastmcp-tasks 패키지가 별도로 필요해요. mcp-tasks 옵션 그룹으로 사용 가능: pip install "pydantic-ai-slim[mcp-tasks]".
FastMCP 3 서버의 경우 pip install "fastmcp[tasks]>=3,<4"로 tasks extra를 설치하고 task=TaskConfig(mode=...)로 도구별 task 지원을 선언하세요:
background_task_server.py
from fastmcp import FastMCP
from fastmcp.server.tasks import TaskConfig
mcp = FastMCP('long_running_server')
@mcp.tool(task=TaskConfig(mode='optional'))
async def deep_research(topic: str) -> str:
import asyncio
await asyncio.sleep(0)
return f'Researched {topic}'
if __name__ == '__main__':
mcp.run(transport='streamable-http')
기본적으로 MCPToolset은 도구가 지원할 때 task-augmented execution을 사용해요. task 지원이 선택적인 도구에 대해 일반 호출을 선호하는 클라이언트는 prefer_tasks=False를 설정할 수 있어요. 이 설정은 task 지원이 필수인 도구에는 영향을 주지 않아요:
background_task_client.py
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
toolset = MCPToolset('http://localhost:8000/mcp', prefer_tasks=False)
agent = Agent('openai:gpt-5.2', toolsets=[toolset])
Resources
MCP 서버는 resources — 클라이언트가 접근할 수 있는 파일, 데이터, 또는 콘텐츠 — 를 제공할 수 있어요. MCP의 리소스는 애플리케이션 주도적이며, 호스트 애플리케이션이 자체 필요에 따라 컨텍스트를 수동으로 통합하는 방법을 결정해요. 도구가 ResourceLink나 EmbeddedResource를 반환하지 않는 한 LLM에 자동으로 노출되지 않아요.
MCPToolset은 리소스를 발견하고 읽는 메서드를 노출해요:
list_resources()-- 서버의 모든 사용 가능한 리소스 나열list_resource_templates()-- 파라미터 자리 표시자가 있는 리소스 템플릿 나열read_resource(uri)-- URI로 특정 리소스의 콘텐츠 읽기
텍스트 콘텐츠는 str로, 바이너리 콘텐츠는 BinaryContent로 반환돼요.
리소스를 소비하기 전에 일부를 노출하는 서버를 실행해야 해요:
mcp_resource_server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP('Pydantic AI MCP Server')
@mcp.resource('resource://user_name.txt', mime_type='text/plain')
async def user_name_resource() -> str:
return 'Alice'
if __name__ == '__main__':
mcp.run()
그런 다음 클라이언트에서 읽을 수 있어요:
mcp_resources.py
import asyncio
from fastmcp.client.transports import StdioTransport
from pydantic_ai.mcp import MCPToolset
async def main():
toolset = MCPToolset(StdioTransport(command='python', args=['-m', 'mcp_resource_server']))
async with toolset:
# List all available resources
resources = await toolset.list_resources()
for resource in resources:
print(f' - {resource.name}: {resource.uri} ({resource.mime_type})')
#> - user_name_resource: resource://user_name.txt (text/plain)
# Read a text resource
user_name = await toolset.read_resource('resource://user_name.txt')
print(f'Text content: {user_name}')
#> Text content: Alice
if __name__ == '__main__':
asyncio.run(main())
(이 예시는 완전한 코드로, "그대로" 실행할 수 있어요)
HTTP authentication
HTTP transport의 경우 MCPToolset은 auth 인자를 받아들여요. bearer 토큰 문자열, 어떤 httpx2.Auth(또는 FastMCP 3의 레거시 httpx.Auth), 또는 FastMCP의 OAuth 흐름을 활성화하는 리터럴 문자열 'oauth'. API 키 같은 정적 헤더는 headers 인자로 전달할 수 있어요.
Per-user authentication
다중 사용자 또는 다중 테넌트 애플리케이션에서 각 사용자는 보통 테넌트 범위 bearer 토큰 같은 MCP 서버에 대한 자체 자격 증명을 가져요.
공유된 MCPToolset 인스턴스는 단일 정체성이에요
MCPToolset 인스턴스는 그것을 사용하는 모든 동시 에이전트 실행이 공유하는 하나의 MCP 세션을 유지해요. 연결은(인증도) 그것을 가장 먼저 필요로 하는 실행이 확립하고, 마지막 실행이 끝날 때만 해체돼요. auth 객체 안의 ContextVar 같은 task 로컬 상태에서 요청별로 자격 증명을 파생하는 것은 공유 인스턴스에서 동작하지 않아요. 겹치는 실행이 세션을 연 실행의 자격 증명으로 조용히 요청을 보낼 거예요.
해당 사용자의 자격 증명으로 요청을 하려면, 각 동시 실행이 자체 MCPToolset 인스턴스를 필요로 해서 자체 인증된 세션을 확립해야 해요. 권장 방법은 @agent.toolset 데코레이터로 toolset을 동적으로 만드는 것이에요. 데코레이트된 함수는 실행 컨텍스트를 받고 실행의 의존성에서 사용자의 자격 증명을 읽을 수 있어요:
mcp_per_user_auth.py
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
from pydantic_ai.mcp import MCPToolset
@dataclass
class UserDeps:
mcp_token: str
agent = Agent('openai:gpt-5.2', deps_type=UserDeps)
@agent.toolset(per_run_step=False) # (1)
def user_mcp_server(ctx: RunContext[UserDeps]) -> MCPToolset:
return MCPToolset('http://localhost:8000/mcp', auth=ctx.deps.mcp_token)
async def main():
result = await agent.run('What is 7 plus 5?', deps=UserDeps(mcp_token='<token>'))
print(result.output)
#> The answer is 12.
per_run_step=False는 각 실행 단계 전에가 아니라 실행당 한 번 toolset을 만들므로, 전체 실행이 단일 MCP 세션을 공유해요.
(이 예시를 실행하려면 asyncio를 import하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
실행별 toolset의 세션이 실행 자체 안에서 확립되므로, ContextVar에 담긴 자격 증명도 이 패턴으로 올바르게 해석돼요 — 하지만 deps를 통해 전달하는 것이 더 명시적이고 task 로컬 상태에 의존하지 않아요.
동적 toolset의 대안으로, 각 요청에 대해 새 MCPToolset을 직접 만들고 에이전트 실행 메서드의 toolsets 인자에 전달할 수 있어요.
Custom TLS / SSL configuration
일부 환경에서는 HTTPS 연결이 확립되는 방식을 조정해야 해요 — 예를 들어 내부 인증 기관을 신뢰하거나, mTLS용 클라이언트 인증서를 제시하거나, (로컬 개발에서만!) 인증서 검증을 완전히 비활성화. MCPToolset은 자체 사전 구성된 httpx2.AsyncClient를 전달할 수 있도록 http_client 파라미터를 노출해요:
mcp_custom_tls_client.py
import ssl
import httpx2
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
# Trust an internal / self-signed CA
ssl_ctx = ssl.create_default_context(cafile='/etc/ssl/private/my_company_ca.pem')
# Optional: load a client certificate for mutual TLS
ssl_ctx.load_cert_chain(certfile='/etc/ssl/certs/client.crt', keyfile='/etc/ssl/private/client.key')
http_client = httpx2.AsyncClient(verify=ssl_ctx, timeout=httpx2.Timeout(10.0))
toolset = MCPToolset('http://localhost:3001/sse', http_client=http_client) # (1)
agent = Agent('openai:gpt-5.2', toolsets=[toolset])
http_client를 공급하면 Pydantic AI는 모든 요청에 이 클라이언트를 재사용해요. HTTPX가 지원하는 모든 것(verify, cert, 커스텀 프록시, 타임아웃 등)이 따라서 모든 MCP 트래픽에 적용돼요. FastMCP 3에서는 클라이언트를 레거시 httpx로 만드세요.
Client identification
MCP 서버에 연결할 때 Implementation 객체를 클라이언트 정보로 선택적으로 지정해 초기화 중에 서버로 전송할 수 있어요. 이는 다음에 유용해요:
- 서버 로그에서 애플리케이션 식별
- 클라이언트에 따라 커스텀 동작을 제공하도록 서버 허용
- MCP 연결 디버깅 및 모니터링
- 버전 특정 기능 협상
mcp_client_with_name.py
from mcp import types as mcp_types
from pydantic_ai.mcp import MCPToolset
toolset = MCPToolset(
'http://localhost:3001/sse',
client_info=mcp_types.Implementation(
name='MyApplication',
version='2.1.0',
),
)
MCP sampling
MCP sampling이란 무엇인가요?
MCP에서 sampling은 MCP 서버가 MCP 클라이언트를 통해 LLM 호출을 할 수 있게 하는 시스템이에요 — 효과적으로 사용 중인 어떤 transport든 통해 클라이언트를 거쳐 LLM에 요청을 프록시하는 것이죠.
Sampling은 MCP 서버가 Gen AI를 사용해야 하지만 각각 자체 LLM 자격 증명을 제공하고 싶지 않을 때, 또는 공개 MCP 서버가 연결하는 클라이언트가 LLM 호출 비용을 지불하기를 원할 때 매우 유용해요.
혼란스럽게도, 그것은 관찰 가능성의 "sampling" 개념이나, 솔직히 어떤 다른 도메인의 "sampling" 개념과도 관련이 없어요.
Sampling diagram
데이터 흐름을 명확히 할 수도 아닐 수도 있는 mermaid 다이어그램이에요:
sequenceDiagram
participant LLM
participant MCP_Client as MCP client
participant MCP_Server as MCP server
MCP_Client->>LLM: LLM call
LLM->>MCP_Client: LLM tool call response
MCP_Client->>MCP_Server: tool call
MCP_Server->>MCP_Client: sampling "create message"
MCP_Client->>LLM: LLM call
LLM->>MCP_Client: LLM text response
MCP_Client->>MCP_Server: sampling response
MCP_Server->>MCP_Client: tool call response
Pydantic AI는 클라이언트와 서버로서 sampling을 지원해요. 서버 안에서 sampling을 사용하는 방법에 대한 자세한 내용은 server 문서를 참고하세요.
클라이언트로 sampling을 사용하려면 MCPToolset에 sampling_model이 설정되어 있어야 해요. 이는 sampling_model= 생성자 키워드 인자로 toolset에 직접 설정하거나, agent.set_mcp_sampling_model()으로 에이전트의 모델(또는 인자로 지정된 것)을 에이전트에 등록된 모든 MCPToolset의 sampling 모델로 사용할 수 있어요.
sampling을 사용하고 싶은 MCP 서버가 있다고 해봅시다(이 경우 도구 인자에 따라 SVG 생성):
Sampling MCP server
generate_svg.py
import re
from pathlib import Path
from mcp import SamplingMessage
from mcp.server.fastmcp import Context, FastMCP
from mcp.types import TextContent
app = FastMCP()
@app.tool()
async def image_generator(ctx: Context, subject: str, style: str) -> str:
prompt = f'{subject=} {style=}'
# `ctx.session.create_message` is the sampling call
result = await ctx.session.create_message(
[SamplingMessage(role='user', content=TextContent(type='text', text=prompt))],
max_tokens=1_024,
system_prompt='Generate an SVG image as per the user input',
)
assert isinstance(result.content, TextContent)
path = Path(f'{subject}_{style}.svg')
# remove triple backticks if the svg was returned within markdown
if m := re.search(r'^```\w*$(.+?)```$', result.content.text, re.S | re.M):
path.write_text(m.group(1), encoding='utf-8')
else:
path.write_text(result.content.text, encoding='utf-8')
return f'See {path}'
if __name__ == '__main__':
# run the server via stdio
app.run()
Agent와 함께 이 서버를 사용하면 automatially sampling이 허용돼요:
sampling_mcp_client.py
from fastmcp.client.transports import StdioTransport
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
toolset = MCPToolset(StdioTransport(command='python', args=['generate_svg.py']))
agent = Agent('openai:gpt-5.2', toolsets=[toolset])
async def main():
agent.set_mcp_sampling_model()
result = await agent.run('Create an image of a robot in a punk style.')
print(result.output)
#> Image file written to robot_punk.svg.
(이 예시는 완전한 코드로, "그대로" 실행할 수 있어요)
Elicitation
MCP에서 elicitation은 서버가 세션 중 누락되었거나 추가 컨텍스트를 위해 클라이언트에게 구조화된 입력을 요청하게 해요.
Elicitation은 모델이 "잠깐 — 계속하기 전에 X를 알아야 해"라고 본질적으로 말할 수 있게 해주며, 모든 것을 미리 요구하거나 감으로 찍지 않게 해요.
How elicitation works
Elicitation은 ElicitRequest라는 프로토콜 메시지 타입을 도입하며, 추가 정보가 필요할 때 서버가 클라이언트로 보내요. 클라이언트는 그런 다음 ElicitResult 또는 ErrorData 메시지로 응답할 수 있어요.
전형적인 상호작용은 이렇게 생겼어요:
- 사용자가 MCP 서버에 요청(예: "그 이탈리아 식당에 자리 예약해줘")
- 서버는 더 많은 정보가 필요하다는 것을 식별(예: "어느 이탈리아 식당?", "언제?")
- 서버는 누락된 정보를 요청하는
ElicitRequest를 클라이언트로 보내요. - 클라이언트가 요청을 받아 사용자에게 제시(예: 터미널 프롬프트, GUI 대화상자, 웹 인터페이스를 통해).
- 사용자가 요청된 정보를 제공하거나, 거절하거나, 취소해요.
- 클라이언트가 사용자의 응답으로
ElicitResult를 서버에 다시 보내요. - 구조화된 데이터로 서버는 원래 요청 처리를 계속할 수 있어요.
이를 통해 특히 다단계 워크플로우에서 더 상호작용적이고 사용자 친화적인 경험이 가능해져요. 모든 정보를 미리 요구하는 대신, 서버가 필요할 때 물어볼 수 있어요.
Setting up elicitation
elicitation을 활성화하려면 MCPToolset을 만들 때 elicitation_handler를 제공하세요:
restaurant_server.py
from mcp.server.fastmcp import Context, FastMCP
from pydantic import BaseModel, Field
mcp = FastMCP(name='Restaurant Booking')
class BookingDetails(BaseModel):
"""Schema for restaurant booking information."""
restaurant: str = Field(description='Choose a restaurant')
party_size: int = Field(description='Number of people', ge=1, le=8)
date: str = Field(description='Reservation date (DD-MM-YYYY)')
@mcp.tool()
async def book_table(ctx: Context) -> str:
"""Book a restaurant table with user input."""
# Ask user for booking details using Pydantic schema
result = await ctx.elicit(message='Please provide your booking details:', schema=BookingDetails)
if result.action == 'accept' and result.data:
booking = result.data
return f'✅ Booked table for {booking.party_size} at {booking.restaurant} on {booking.date}'
elif result.action == 'decline':
return 'No problem! Maybe another time.'
else: # cancel
return 'Booking cancelled.'
if __name__ == '__main__':
mcp.run(transport='stdio')
이 서버는 book_table 도구가 호출될 때 클라이언트에서 구조화된 예약 세부사항을 요청함으로써 elicitation을 보여줘요. 일치하는 클라이언트를 연결하는 방법은 다음과 같아요:
client_example.py
import asyncio
from fastmcp.client.transports import StdioTransport
from mcp.types import ElicitRequestParams, ElicitResult
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
async def handle_elicitation(context, params: ElicitRequestParams) -> ElicitResult:
"""Handle elicitation requests from MCP server."""
print(f'\n{params.message}')
if not params.requestedSchema:
response = input('Response: ')
return ElicitResult(action='accept', content={'response': response})
# Collect data for each field
properties = params.requestedSchema['properties']
data = {}
for field, info in properties.items():
description = info.get('description', field)
value = input(f'{description}: ')
# Convert to proper type based on JSON schema
if info.get('type') == 'integer':
data[field] = int(value)
else:
data[field] = value
# Confirm
confirm = input('\nConfirm booking? (y/n/c): ').lower()
if confirm == 'y':
print('Booking details:', data)
return ElicitResult(action='accept', content=data)
elif confirm == 'n':
return ElicitResult(action='decline')
else:
return ElicitResult(action='cancel')
toolset = MCPToolset(
StdioTransport(command='python', args=['restaurant_server.py']),
elicitation_handler=handle_elicitation,
)
agent = Agent('openai:gpt-5.2', toolsets=[toolset])
async def main():
"""Run the agent to book a restaurant table."""
result = await agent.run('Book me a table')
print(f'\nResult: {result.output}')
if __name__ == '__main__':
asyncio.run(main())
Supported schema types
MCP elicitation은 문자열, 숫자, 불리언, enum 타입만 평평한 객체 구조로 지원해요. 이러한 제한으로 신뢰할 수 있는 크로스 클라이언트 호환성을 보장해요. 자세한 내용은 supported schema types를 참고하세요.
Security
MCP elicitation은 신중한 처리가 필요해요. 서버는 민감한 정보를 요청해서는 안 되며, 클라이언트는 명확한 설명과 함께 사용자 승인 제어를 구현해야 해요. 자세한 내용은 security considerations를 참고하세요.