Expose MCP server instructions in MCPServer.instructions property (#3431)

Co-authored-by: Motta Kin <[email protected]>
Co-authored-by: Douwe Maan <[email protected]>

Author: 3 people

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.

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)
 

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'
 
 

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
 

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.')