Load MCP servers from file (#2698)

Author: Kludex

docs/mcp/client.md

@@ -163,6 +163,59 @@ async def main():
 
 1. See [MCP Run Python](https://github.com/pydantic/mcp-run-python) for more information.
 
+## Loading MCP Servers from Configuration
+
+Instead of creating MCP server instances individually in code, you can load multiple servers from a JSON configuration file using [`load_mcp_servers()`][pydantic_ai.mcp.load_mcp_servers].
+
+This is particularly useful when you need to manage multiple MCP servers or want to configure servers externally without modifying code.
+
+### Configuration Format
+
+The configuration file should be a JSON file with an `mcpServers` object containing server definitions. Each server is identified by a unique key and contains the configuration for that server type:
+
+```json {title="mcp_config.json"}
+{
+  "mcpServers": {
+    "python-runner": {
+      "command": "uv",
+      "args": ["run", "mcp-run-python", "stdio"]
+    },
+    "weather-api": {
+      "url": "http://localhost:3001/sse"
+    },
+    "calculator": {
+      "url": "http://localhost:8000/mcp"
+    }
+  }
+}
+```
+
+!!! note
+    The MCP server is only inferred to be an SSE server because of the `/sse` suffix.
+    Any other server with the "url" field will be inferred to be a Streamable HTTP server.
+
+    We made this decision given that the SSE transport is deprecated.
+
+### Usage
+
+```python {title="mcp_config_loader.py" test="skip"}
+from pydantic_ai import Agent
+from pydantic_ai.mcp import load_mcp_servers
+
+# Load all servers from configuration file
+servers = load_mcp_servers('mcp_config.json')
+
+# Create agent with all loaded servers
+agent = Agent('openai:gpt-5', toolsets=servers)
+
+async def main():
+    async with agent:
+        result = await agent.run('What is 7 plus 5?')
+    print(result.output)
+```
+
+_(This example is complete, it can be run "as is" — you'll need to add `asyncio.run(main())` to run `main`)_
+
 ## Tool call customisation
 
 The MCP servers provide the ability to set a `process_tool_call` which allows

pydantic_ai_slim/pydantic_ai/mcp.py

@@ -10,12 +10,14 @@
 from dataclasses import field, replace
 from datetime import timedelta
 from pathlib import Path
-from typing import Any
+from typing import Annotated, Any
 
 import anyio
 import httpx
 import pydantic_core
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+from pydantic import BaseModel, Discriminator, Field, Tag
+from pydantic_core import CoreSchema, core_schema
 from typing_extensions import Self, assert_never, deprecated
 
 from pydantic_ai.tools import RunContext, ToolDefinition
@@ -41,7 +43,7 @@
 # after mcp imports so any import error maps to this file, not _mcp.py
 from . import _mcp, _utils, exceptions, messages, models
 
-__all__ = 'MCPServer', 'MCPServerStdio', 'MCPServerHTTP', 'MCPServerSSE', 'MCPServerStreamableHTTP'
+__all__ = 'MCPServer', 'MCPServerStdio', 'MCPServerHTTP', 'MCPServerSSE', 'MCPServerStreamableHTTP', 'load_mcp_servers'
 
 TOOL_SCHEMA_VALIDATOR = pydantic_core.SchemaValidator(
     schema=pydantic_core.core_schema.dict_schema(
@@ -498,6 +500,22 @@ def __init__(
             id=id,
         )
 
+    @classmethod
+    def __get_pydantic_core_schema__(cls, _: Any, __: Any) -> CoreSchema:
+        return core_schema.no_info_after_validator_function(
+            lambda dct: MCPServerStdio(**dct),
+            core_schema.typed_dict_schema(
+                {
+                    'command': core_schema.typed_dict_field(core_schema.str_schema()),
+                    'args': core_schema.typed_dict_field(core_schema.list_schema(core_schema.str_schema())),
+                    'env': core_schema.typed_dict_field(
+                        core_schema.dict_schema(core_schema.str_schema(), core_schema.str_schema()),
+                        required=False,
+                    ),
+                }
+            ),
+        )
+
     @asynccontextmanager
     async def client_streams(
         self,
@@ -520,6 +538,16 @@ def __repr__(self) -> str:
             repr_args.append(f'id={self.id!r}')
         return f'{self.__class__.__name__}({", ".join(repr_args)})'
 
+    def __eq__(self, value: object, /) -> bool:
+        if not isinstance(value, MCPServerStdio):
+            return False  # pragma: no cover
+        return (
+            self.command == value.command
+            and self.args == value.args
+            and self.env == value.env
+            and self.cwd == value.cwd
+        )
+
 
 class _MCPServerHTTP(MCPServer):
     url: str
@@ -733,10 +761,29 @@ async def main():
     1. This will connect to a server running on `localhost:3001`.
     """
 
+    @classmethod
+    def __get_pydantic_core_schema__(cls, _: Any, __: Any) -> CoreSchema:
+        return core_schema.no_info_after_validator_function(
+            lambda dct: MCPServerSSE(**dct),
+            core_schema.typed_dict_schema(
+                {
+                    'url': core_schema.typed_dict_field(core_schema.str_schema()),
+                    'headers': core_schema.typed_dict_field(
+                        core_schema.dict_schema(core_schema.str_schema(), core_schema.str_schema()), required=False
+                    ),
+                }
+            ),
+        )
+
     @property
     def _transport_client(self):
         return sse_client  # pragma: no cover
 
+    def __eq__(self, value: object, /) -> bool:
+        if not isinstance(value, MCPServerSSE):
+            return False  # pragma: no cover
+        return self.url == value.url
+
 
 @deprecated('The `MCPServerHTTP` class is deprecated, use `MCPServerSSE` instead.')
 class MCPServerHTTP(MCPServerSSE):
@@ -790,10 +837,29 @@ async def main():
     ```
     """
 
+    @classmethod
+    def __get_pydantic_core_schema__(cls, _: Any, __: Any) -> CoreSchema:
+        return core_schema.no_info_after_validator_function(
+            lambda dct: MCPServerStreamableHTTP(**dct),
+            core_schema.typed_dict_schema(
+                {
+                    'url': core_schema.typed_dict_field(core_schema.str_schema()),
+                    'headers': core_schema.typed_dict_field(
+                        core_schema.dict_schema(core_schema.str_schema(), core_schema.str_schema()), required=False
+                    ),
+                }
+            ),
+        )
+
     @property
     def _transport_client(self):
         return streamablehttp_client  # pragma: no cover
 
+    def __eq__(self, value: object, /) -> bool:
+        if not isinstance(value, MCPServerStreamableHTTP):
+            return False  # pragma: no cover
+        return self.url == value.url
+
 
 ToolResult = (
     str
@@ -823,3 +889,50 @@ def _transport_client(self):
 Allows wrapping an MCP server tool call to customize it, including adding extra request
 metadata.
 """
+
+
+def _mcp_server_discriminator(value: dict[str, Any]) -> str | None:
+    if 'url' in value:
+        if value['url'].endswith('/sse'):
+            return 'sse'
+        return 'streamable-http'
+    return 'stdio'
+
+
+class MCPServerConfig(BaseModel):
+    """Configuration for MCP servers."""
+
+    mcp_servers: Annotated[
+        dict[
+            str,
+            Annotated[
+                Annotated[MCPServerStdio, Tag('stdio')]
+                | Annotated[MCPServerStreamableHTTP, Tag('streamable-http')]
+                | Annotated[MCPServerSSE, Tag('sse')],
+                Discriminator(_mcp_server_discriminator),
+            ],
+        ],
+        Field(alias='mcpServers'),
+    ]
+
+
+def load_mcp_servers(config_path: str | Path) -> list[MCPServerStdio | MCPServerStreamableHTTP | MCPServerSSE]:
+    """Load MCP servers from a configuration file.
+
+    Args:
+        config_path: The path to the configuration file.
+
+    Returns:
+        A list of MCP servers.
+
+    Raises:
+        FileNotFoundError: If the configuration file does not exist.
+        ValidationError: If the configuration file does not match the schema.
+    """
+    config_path = Path(config_path)
+
+    if not config_path.exists():
+        raise FileNotFoundError(f'Config file {config_path} not found')
+
+    config = MCPServerConfig.model_validate_json(config_path.read_bytes())
+    return list(config.mcp_servers.values())

tests/models/cassettes/test_model_names/test_known_model_names.yaml

@@ -99,19 +99,53 @@ interactions:
       alt-svc:
       - h3=":443"; ma=86400
       content-length:
-      - '99'
+      - '762'
       content-type:
       - application/json
       referrer-policy:
       - strict-origin-when-cross-origin
       strict-transport-security:
       - max-age=3600; includeSubDomains
     parsed_body:
-      code: wrong_api_key
-      message: Wrong API Key
-      param: api_key
-      type: invalid_request_error
+      data:
+      - created: 0
+        id: llama3.1-8b
+        object: model
+        owned_by: Cerebras
+      - created: 0
+        id: qwen-3-coder-480b
+        object: model
+        owned_by: Cerebras
+      - created: 0
+        id: llama-4-maverick-17b-128e-instruct
+        object: model
+        owned_by: Cerebras
+      - created: 0
+        id: qwen-3-235b-a22b-thinking-2507
+        object: model
+        owned_by: Cerebras
+      - created: 0
+        id: llama-3.3-70b
+        object: model
+        owned_by: Cerebras
+      - created: 0
+        id: qwen-3-32b
+        object: model
+        owned_by: Cerebras
+      - created: 0
+        id: qwen-3-235b-a22b-instruct-2507
+        object: model
+        owned_by: Cerebras
+      - created: 0
+        id: llama-4-scout-17b-16e-instruct
+        object: model
+        owned_by: Cerebras
+      - created: 0
+        id: gpt-oss-120b
+        object: model
+        owned_by: Cerebras
+      object: list
     status:
-      code: 401
-      message: Unauthorized
+      code: 200
+      message: OK
 version: 1

tests/test_mcp.py

@@ -14,6 +14,7 @@
 
 from pydantic_ai.agent import Agent
 from pydantic_ai.exceptions import ModelRetry, UnexpectedModelBehavior, UserError
+from pydantic_ai.mcp import MCPServerStreamableHTTP, load_mcp_servers
 from pydantic_ai.messages import (
     BinaryContent,
     ModelRequest,
@@ -1375,3 +1376,19 @@ async def test_elicitation_callback_not_set(run_context: RunContext[int]):
         # Should raise an error when elicitation is attempted without callback
         with pytest.raises(ModelRetry, match='Elicitation not supported'):
             await server.direct_call_tool('use_elicitation', {'question': 'Should I continue?'})
+
+
+def test_load_mcp_servers(tmp_path: Path):
+    config = tmp_path / 'mcp.json'
+
+    config.write_text('{"mcpServers": {"potato": {"url": "https://example.com/mcp"}}}')
+    assert load_mcp_servers(config) == snapshot([MCPServerStreamableHTTP(url='https://example.com/mcp')])
+
+    config.write_text('{"mcpServers": {"potato": {"command": "python", "args": ["-m", "tests.mcp_server"]}}}')
+    assert load_mcp_servers(config) == snapshot([MCPServerStdio(command='python', args=['-m', 'tests.mcp_server'])])
+
+    config.write_text('{"mcpServers": {"potato": {"url": "https://example.com/sse"}}}')
+    assert load_mcp_servers(config) == snapshot([MCPServerSSE(url='https://example.com/sse')])
+
+    with pytest.raises(FileNotFoundError):
+        load_mcp_servers(tmp_path / 'does_not_exist.json')