Server
Server
이 문서에서는 Pydantic AI 모델을 MCP 서버 안에서 사용하는 방법을 알려드려요. 도구 호출 안에서 Pydantic AI 에이전트를 실행하는 간단한 MCP 서버 예시부터, MCP sampling을 활용해 MCP 클라이언트를 통해 LLM 호출을 하는 확장 예시까지 살펴볼 수 있어요.
출처: 문서
본문
Pydantic AI 모델은 MCP 서버 안에서도 사용할 수 있어요.
MCP Server
도구 호출 안에서 Pydantic AI를 사용하는 간단한 Python MCP 서버의 예시예요:
mcp_server.py
from mcp.server.fastmcp import FastMCP
from pydantic_ai import Agent
server = FastMCP('Pydantic AI Server')
server_agent = Agent(
'anthropic:claude-haiku-4-5', instructions='always reply in rhyme'
)
@server.tool()
async def poet(theme: str) -> str:
"""Poem generator"""
r = await server_agent.run(f'write a poem about {theme}')
return r.output
if __name__ == '__main__':
server.run()
Simple client
이 서버는 어떤 MCP 클라이언트로도 쿼리할 수 있어요. 다음은 Python SDK를 직접 사용한 예시예요:
mcp_client.py
import asyncio
import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def client():
server_params = StdioServerParameters(
command='python', args=['mcp_server.py'], env=os.environ
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
result = await session.call_tool('poet', {'theme': 'socks'})
print(result.content[0].text)
"""
Oh, socks, those garments soft and sweet,
That nestle softly 'round our feet,
From cotton, wool, or blended thread,
They keep our toes from feeling dread.
"""
if __name__ == '__main__':
asyncio.run(client())
MCP Sampling
MCP Sampling이란 무엇인가요?
MCP sampling이 무엇이고, Pydantic AI를 MCP 클라이언트로 사용할 때 이를 어떻게 지원할 수 있는지에 대한 자세한 내용은 MCP 클라이언트 문서를 참고하세요.
Pydantic AI 에이전트가 MCP 서버 안에서 사용될 때, MCPSamplingModel을 통해 sampling을 사용할 수 있어요.
다른 모델에서 이어서 대화를 계속하려면 해당 모델의 message_history를 sampling 에이전트에 전달하면 돼요. 해당 히스토리의 함수 도구 호출, 결과, 재시도 피드백은 MCP의 네이티브 도구 콘텐츠 블록을 사용해 보존되는데, 이를 위해서는 2025-11-25 sampling 형식 이상을 지원하는 클라이언트가 필요해요. 멀티모달 도구 결과는 아직 지원되지 않아요. 도구 히스토리를 재생한다고 해서 sampling 에이전트가 새 도구를 호출할 수 있게 되지는 않아요.
위 예시를 확장해 sampling을 사용할 수 있는데, LLM에 직접 연결하는 대신 에이전트가 MCP 클라이언트를 통해 콜백해서 LLM 호출을 하는 방식이에요.
mcp_server_sampling.py
from mcp.server.fastmcp import Context, FastMCP
from pydantic_ai import Agent
from pydantic_ai.models.mcp_sampling import MCPSamplingModel
server = FastMCP('Pydantic AI Server with sampling')
server_agent = Agent(instructions='always reply in rhyme')
@server.tool()
async def poet(ctx: Context, theme: str) -> str:
"""Poem generator"""
r = await server_agent.run(f'write a poem about {theme}', model=MCPSamplingModel(session=ctx.session))
return r.output
if __name__ == '__main__':
server.run() # run the server over stdio
위의 클라이언트는 sampling을 지원하지 않으므로, 이 서버와 함께 사용하려 하면 오류를 받게 돼요.
MCP 클라이언트에서 sampling을 지원하는 가장 간단한 방법은 Pydantic AI 에이전트를 클라이언트로 사용하는 것이지만, vanilla MCP SDK로 sampling을 지원하려면 다음과 같이 할 수 있어요:
mcp_client_sampling.py
import asyncio
from typing import Any
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.shared.context import RequestContext
from mcp.types import (
CreateMessageRequestParams,
CreateMessageResult,
ErrorData,
TextContent,
)
async def sampling_callback(
context: RequestContext[ClientSession, Any], params: CreateMessageRequestParams
) -> CreateMessageResult | ErrorData:
print('sampling system prompt:', params.systemPrompt)
#> sampling system prompt: always reply in rhyme
print('sampling messages:', params.messages)
"""
sampling messages:
[
SamplingMessage(
role='user',
content=TextContent(
type='text',
text='write a poem about socks',
annotations=None,
meta=None,
),
meta=None,
)
]
"""
# TODO get the response content by calling an LLM...
response_content = 'Socks for a fox.'
return CreateMessageResult(
role='assistant',
content=TextContent(type='text', text=response_content),
model='fictional-llm',
)
async def client():
server_params = StdioServerParameters(command='python', args=['mcp_server_sampling.py'])
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write, sampling_callback=sampling_callback) as session:
await session.initialize()
result = await session.call_tool('poet', {'theme': 'socks'})
print(result.content[0].text)
#> Socks for a fox.
if __name__ == '__main__':
asyncio.run(client())
(이 예시는 완전한 코드로, "그대로" 실행할 수 있어요)