1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
# Server
PydanticAI models can also be used within MCP Servers.
## MCP Server
Here's a simple example of a [Python MCP server](https://github.com/modelcontextprotocol/python-sdk) using PydanticAI within a tool call:
```py {title="mcp_server.py" py="3.10"}
from mcp.server.fastmcp import FastMCP
from pydantic_ai import Agent
server = FastMCP('PydanticAI Server')
server_agent = Agent(
'anthropic:claude-3-5-haiku-latest', system_prompt='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
This server can be queried with any MCP client. Here is an example using the Python SDK directly:
```py {title="mcp_client.py" py="3.10" requires="mcp_server.py" dunder_name="not_main"}
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
!!! info "What is MCP Sampling?"
See the [MCP client docs](./client.md#mcp-sampling) for details of what MCP sampling is, and how you can support it when using Pydantic AI as an MCP client.
When Pydantic AI agents are used within MCP servers, they can use sampling via [`MCPSamplingModel`][pydantic_ai.models.mcp_sampling.MCPSamplingModel].
We can extend the above example to use sampling so instead of connecting directly to the LLM, the agent calls back through the MCP client to make LLM calls.
```py {title="mcp_server_sampling.py" py="3.10"}
from mcp.server.fastmcp import Context, FastMCP
from pydantic_ai import Agent
from pydantic_ai.models.mcp_sampling import MCPSamplingModel
server = FastMCP('PydanticAI Server with sampling')
server_agent = Agent(system_prompt='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
```
The [above](#simple-client) client does not support sampling, so if you tried to use it with this server you'd get an error.
The simplest way to support sampling in an MCP client is to [use](./client.md#mcp-sampling) a Pydantic AI agent as the client, but if you wanted to support sampling with the vanilla MCP SDK, you could do so like this:
```py {title="mcp_client_sampling.py" py="3.10" requires="mcp_server_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,
),
)
]
"""
# 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())
```
_(This example is complete, it can be run "as is" with Python 3.10+)_
|