diff --git a/docs/mcp/client.md b/docs/mcp/client.md index 71e15b77a2..39fd80bc4d 100644 --- a/docs/mcp/client.md +++ b/docs/mcp/client.md @@ -338,6 +338,29 @@ calculator_server = MCPServerSSE( agent = Agent('openai:gpt-5', toolsets=[weather_server, calculator_server]) ``` +## Server Instructions + +MCP servers can provide instructions during initialization that give context about how to best interact with the server's tools. These instructions are accessible via the [`instructions`][pydantic_ai.mcp.MCPServer.instructions] property after the server connection is established. + +```python {title="mcp_server_instructions.py"} +from pydantic_ai import Agent +from pydantic_ai.mcp import MCPServerStreamableHTTP + +server = MCPServerStreamableHTTP('http://localhost:8000/mcp') +agent = Agent('openai:gpt-5', toolsets=[server]) + +@agent.instructions +async def mcp_server_instructions(): + return server.instructions # (1)! + +async def main(): + result = await agent.run('What is 7 plus 5?') + print(result.output) + #> The answer is 12. +``` + +1. The server connection is guaranteed to be established by this point, so `server.instructions` is available. + ## Tool metadata MCP tools can include metadata that provides additional information about the tool's characteristics, which can be useful when [filtering tools][pydantic_ai.toolsets.FilteredToolset]. The `meta`, `annotations`, and `output_schema` fields can be found on the `metadata` dict on the [`ToolDefinition`][pydantic_ai.tools.ToolDefinition] object that's passed to filter functions. diff --git a/pydantic_ai_slim/pydantic_ai/mcp.py b/pydantic_ai_slim/pydantic_ai/mcp.py index 3d2b607b3f..982847bacf 100644 --- a/pydantic_ai_slim/pydantic_ai/mcp.py +++ b/pydantic_ai_slim/pydantic_ai/mcp.py @@ -122,6 +122,7 @@ class MCPServer(AbstractToolset[Any], ABC): _read_stream: MemoryObjectReceiveStream[SessionMessage | Exception] _write_stream: MemoryObjectSendStream[SessionMessage] _server_info: mcp_types.Implementation + _instructions: str | None def __init__( self, @@ -200,6 +201,15 @@ def server_info(self) -> mcp_types.Implementation: ) return self._server_info + @property + def instructions(self) -> str | None: + """Access the instructions sent by the MCP server during initialization.""" + if not hasattr(self, '_instructions'): + raise AttributeError( + f'The `{self.__class__.__name__}.instructions` is only available after initialization.' + ) + return self._instructions + async def list_tools(self) -> list[mcp_types.Tool]: """Retrieve tools that are currently active on the server. @@ -337,6 +347,7 @@ async def __aenter__(self) -> Self: with anyio.fail_after(self.timeout): result = await self._client.initialize() self._server_info = result.serverInfo + self._instructions = result.instructions if log_level := self.log_level: await self._client.set_logging_level(log_level) diff --git a/tests/mcp_server.py b/tests/mcp_server.py index 54b105ab29..eaec8eb9a2 100644 --- a/tests/mcp_server.py +++ b/tests/mcp_server.py @@ -16,7 +16,7 @@ ) from pydantic import AnyUrl, BaseModel -mcp = FastMCP('Pydantic AI MCP Server') +mcp = FastMCP('Pydantic AI MCP Server', instructions='Be a helpful assistant.') log_level = 'unset' diff --git a/tests/test_examples.py b/tests/test_examples.py index 85bae688d0..407816b60a 100644 --- a/tests/test_examples.py +++ b/tests/test_examples.py @@ -304,6 +304,10 @@ class MockMCPServer(AbstractToolset[Any]): def id(self) -> str | None: return None # pragma: no cover + @property + def instructions(self) -> str | None: + return None + async def __aenter__(self) -> MockMCPServer: return self diff --git a/tests/test_mcp.py b/tests/test_mcp.py index 9d0654f9df..2f05f419a6 100644 --- a/tests/test_mcp.py +++ b/tests/test_mcp.py @@ -1757,6 +1757,15 @@ async def test_server_info(mcp_server: MCPServerStdio) -> None: assert mcp_server.server_info.name == 'Pydantic AI MCP Server' +async def test_instructions(mcp_server: MCPServerStdio) -> None: + with pytest.raises( + AttributeError, match='The `MCPServerStdio.instructions` is only available after initialization.' + ): + mcp_server.instructions + async with mcp_server: + assert mcp_server.instructions == 'Be a helpful assistant.' + + async def test_agent_run_stream_with_mcp_server_http(allow_model_requests: None, model: Model): server = MCPServerStreamableHTTP(url='https://mcp.deepwiki.com/mcp', timeout=30) agent = Agent(model, toolsets=[server], instructions='Be concise.')