Skip to content

Commit 7d20254

Browse files
committed
First cut at config API
1 parent 1eb1bba commit 7d20254

File tree

3 files changed

+125
-0
lines changed

3 files changed

+125
-0
lines changed
Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
"""Configuration management for MCP servers."""
2+
3+
# stdlib imports
4+
import json
5+
from pathlib import Path
6+
from typing import Annotated, Any, Literal
7+
8+
# third party imports
9+
from pydantic import BaseModel, Field, model_validator
10+
11+
12+
class MCPServerConfig(BaseModel):
13+
"""Base class for MCP server configurations."""
14+
15+
pass
16+
17+
18+
class StdioServerConfig(MCPServerConfig):
19+
"""Configuration for stdio-based MCP servers."""
20+
21+
type: Literal["stdio"] = "stdio"
22+
command: str
23+
args: list[str] | None = None
24+
env: dict[str, str] | None = None
25+
26+
27+
class StreamableHttpConfig(MCPServerConfig):
28+
"""Configuration for StreamableHTTP-based MCP servers."""
29+
30+
type: Literal["streamable_http"] = "streamable_http"
31+
url: str
32+
headers: dict[str, str] | None = None
33+
34+
35+
# Discriminated union for different server config types
36+
ServerConfigUnion = Annotated[StdioServerConfig | StreamableHttpConfig, Field(discriminator="type")]
37+
38+
39+
class MCPServersConfig(BaseModel):
40+
"""Configuration for multiple MCP servers."""
41+
42+
servers: dict[str, ServerConfigUnion] = Field(alias="mcpServers")
43+
44+
@model_validator(mode="before")
45+
@classmethod
46+
def infer_server_types(cls, data: Any) -> Any:
47+
"""Automatically infer server types when 'type' field is omitted."""
48+
if isinstance(data, dict) and "mcpServers" in data:
49+
for _server_name, server_config in data["mcpServers"].items(): # type: ignore
50+
if isinstance(server_config, dict) and "type" not in server_config:
51+
# Infer type based on distinguishing fields
52+
if "command" in server_config:
53+
server_config["type"] = "stdio"
54+
elif "url" in server_config:
55+
server_config["type"] = "streamable_http"
56+
return data
57+
58+
@classmethod
59+
def from_file(cls, config_path: Path) -> "MCPServersConfig":
60+
"""Load configuration from a JSON file."""
61+
with open(config_path) as config_file:
62+
return cls.model_validate(json.load(config_file))

tests/client/config/__init__.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
# stdlib imports
2+
import json
3+
from pathlib import Path
4+
5+
# third party imports
6+
import pytest
7+
8+
# local imports
9+
from mcp.client.config.mcp_servers_config import (
10+
MCPServersConfig,
11+
StdioServerConfig,
12+
StreamableHttpConfig,
13+
)
14+
15+
16+
@pytest.fixture
17+
def mcp_config_file(tmp_path: Path) -> Path:
18+
"""Create temporary JSON config file with mixed server types"""
19+
20+
config_data = {
21+
"mcpServers": {
22+
"stdio_server": {
23+
"command": "python",
24+
"args": ["-m", "my_server"],
25+
"env": {"DEBUG": "true"},
26+
},
27+
"http_streamable": {
28+
"url": "https://api.example.com/mcp",
29+
"headers": {"Authorization": "Bearer token123"},
30+
},
31+
}
32+
}
33+
34+
# Write to temporary file
35+
config_file_path = tmp_path / "mcp.json"
36+
with open(config_file_path, "w") as config_file:
37+
json.dump(config_data, config_file)
38+
39+
return config_file_path
40+
41+
42+
def test_stdio_server(mcp_config_file: Path):
43+
config = MCPServersConfig.from_file(mcp_config_file)
44+
45+
stdio_server = config.servers["stdio_server"]
46+
assert isinstance(stdio_server, StdioServerConfig)
47+
48+
assert stdio_server.command == "python"
49+
assert stdio_server.args == ["-m", "my_server"]
50+
assert stdio_server.env == {"DEBUG": "true"}
51+
assert stdio_server.type == "stdio" # Should be automatically inferred
52+
53+
54+
def test_streamable_http_server(mcp_config_file: Path):
55+
config = MCPServersConfig.from_file(mcp_config_file)
56+
57+
http_server = config.servers["http_streamable"]
58+
assert isinstance(http_server, StreamableHttpConfig)
59+
60+
assert http_server.url == "https://api.example.com/mcp"
61+
assert http_server.headers == {"Authorization": "Bearer token123"}
62+
assert http_server.type == "streamable_http" # Should be automatically inferred

0 commit comments

Comments
 (0)