Spaces:
Running
Running
File size: 12,756 Bytes
5f2c09d | 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 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 | """Integration tests for fastmcp.json configuration system."""
import json
import sys
from pathlib import Path
import pytest
from fastmcp.client import Client
from fastmcp.utilities.fastmcp_config import FastMCPConfig
@pytest.fixture
def server_with_config(tmp_path):
"""Create a complete server setup with fastmcp.json config."""
# Create server file
server_file = tmp_path / "server.py"
server_file.write_text("""
from fastmcp import FastMCP
mcp = FastMCP("Config Test Server")
@mcp.tool
def hello(name: str = "World") -> str:
'''Say hello to someone'''
return f"Hello, {name}!"
@mcp.resource("resource://greeting")
def get_greeting() -> str:
'''Get a greeting message'''
return "Welcome to FastMCP!"
if __name__ == "__main__":
import asyncio
asyncio.run(mcp.run_async())
""")
# Create config file
config_data = {
"$schema": "https://gofastmcp.com/schemas/fastmcp_config/v1.json",
"entrypoint": "server.py",
"environment": {
"python": sys.version.split()[0], # Use current Python version
"dependencies": ["fastmcp"],
},
"deployment": {"transport": "stdio", "log_level": "INFO"},
}
config_file = tmp_path / "fastmcp.json"
config_file.write_text(json.dumps(config_data, indent=2))
return tmp_path
class TestConfigFileDetection:
"""Test configuration file detection patterns."""
def test_detect_standard_fastmcp_json(self, tmp_path):
"""Test detection of standard fastmcp.json file."""
config_file = tmp_path / "fastmcp.json"
config_file.write_text(json.dumps({"entrypoint": "server.py"}))
# Should be detected as fastmcp config
assert "fastmcp.json" in config_file.name
assert config_file.name.endswith("fastmcp.json")
def test_detect_prefixed_fastmcp_json(self, tmp_path):
"""Test detection of prefixed fastmcp.json files."""
config_file = tmp_path / "my.fastmcp.json"
config_file.write_text(json.dumps({"entrypoint": "server.py"}))
# Should be detected as fastmcp config
assert "fastmcp.json" in config_file.name
def test_detect_test_fastmcp_json(self, tmp_path):
"""Test detection of test_fastmcp.json file."""
config_file = tmp_path / "test_fastmcp.json"
config_file.write_text(json.dumps({"entrypoint": "server.py"}))
# Should be detected as fastmcp config
assert "fastmcp.json" in config_file.name
class TestConfigWithClient:
"""Test fastmcp.json configuration with client connections."""
@pytest.mark.asyncio
async def test_config_server_with_client(self, server_with_config):
"""Test that a server loaded from config works with a client."""
# Load the config
config_file = server_with_config / "fastmcp.json"
config = FastMCPConfig.from_file(config_file)
# Import the server using the entrypoint
import importlib.util
import sys
entrypoint = config.get_entrypoint(config_file)
spec = importlib.util.spec_from_file_location("test_server", entrypoint.file)
if spec is None or spec.loader is None:
raise RuntimeError(f"Could not load module from {entrypoint.file}")
module = importlib.util.module_from_spec(spec)
sys.modules["test_server"] = module
spec.loader.exec_module(module)
server = module.mcp
# Connect client to server
async with Client(server) as client:
# Test tool
result = await client.call_tool("hello", {"name": "FastMCP"})
assert result.data == "Hello, FastMCP!" # Use .data for string result
# Test resource
results = await client.read_resource("resource://greeting")
assert len(results) == 1
# Resource results should have text content
assert hasattr(results[0], "text") or hasattr(results[0], "contents")
# Get the text content from the resource
text = getattr(results[0], "text", None) or getattr(
results[0], "contents", ""
)
assert "Welcome to FastMCP!" in str(text)
class TestEnvironmentExecution:
"""Test environment configuration execution paths."""
def test_needs_uv_with_dependencies(self):
"""Test that environment with dependencies needs UV."""
config = FastMCPConfig(
entrypoint="server.py",
environment={"dependencies": ["requests", "numpy"]}, # type: ignore[arg-type]
)
assert config.environment is not None
assert config.environment.needs_uv()
def test_needs_uv_with_python_version(self):
"""Test that environment with Python version needs UV."""
config = FastMCPConfig(
entrypoint="server.py",
environment={"python": "3.12"}, # type: ignore[arg-type]
)
assert config.environment is not None
assert config.environment.needs_uv()
def test_no_uv_needed_without_environment(self):
"""Test that no UV is needed without environment config."""
config = FastMCPConfig(entrypoint="server.py")
# Environment is now always present but may be empty
assert config.environment is not None
assert not config.environment.needs_uv()
def test_no_uv_needed_with_empty_environment(self):
"""Test that no UV is needed with empty environment config."""
config = FastMCPConfig(
entrypoint="server.py",
environment={}, # type: ignore[arg-type]
)
assert config.environment is not None
assert not config.environment.needs_uv()
class TestCLIArgumentMerging:
"""Test CLI argument merging with config values."""
def test_cli_overrides_environment(self):
"""Test that CLI args override environment config."""
config = FastMCPConfig(
entrypoint="server.py",
environment={"python": "3.11", "dependencies": ["requests"]}, # type: ignore[arg-type]
)
assert config.environment is not None
merged = config.environment.merge_with_cli_args(
python="3.12", # Override Python version
with_packages=["numpy"], # Add package
with_requirements=None,
project=None,
)
assert merged["python"] == "3.12" # CLI wins
assert "requests" in merged["with_packages"] # From config
assert "numpy" in merged["with_packages"] # From CLI
def test_cli_overrides_deployment(self):
"""Test that CLI args override deployment config."""
config = FastMCPConfig(
entrypoint="server.py",
deployment={"transport": "stdio", "port": 3000, "log_level": "INFO"}, # type: ignore[arg-type]
)
assert config.deployment is not None
merged = config.deployment.merge_with_cli_args(
transport="http", # Override transport
host="localhost", # New value
port=8080, # Override port
path=None,
log_level="DEBUG", # Override log level
server_args=None,
)
assert merged["transport"] == "http" # CLI wins
assert merged["host"] == "localhost" # CLI value
assert merged["port"] == 8080 # CLI wins
assert merged["log_level"] == "DEBUG" # CLI wins
def test_config_values_when_cli_is_none(self):
"""Test that config values are used when CLI args are None."""
config = FastMCPConfig(
entrypoint="server.py",
deployment={"transport": "http", "port": 3000}, # type: ignore[arg-type]
)
assert config.deployment is not None
merged = config.deployment.merge_with_cli_args(
transport=None, # Use config
host=None, # No value
port=None, # Use config
path=None,
log_level=None,
server_args=None,
)
assert merged["transport"] == "http" # From config
assert merged["port"] == 3000 # From config
assert merged["host"] is None # No value provided
class TestPathResolution:
"""Test path resolution in configurations."""
def test_entrypoint_path_resolution(self, tmp_path):
"""Test that entrypoint paths are resolved relative to config."""
# Create nested directory structure
config_dir = tmp_path / "config"
config_dir.mkdir()
src_dir = tmp_path / "src"
src_dir.mkdir()
# Server is in src, config is in config
server_file = src_dir / "server.py"
server_file.write_text("# Server")
config = FastMCPConfig(entrypoint="../src/server.py")
# Get entrypoint resolved relative to config location
config_file = config_dir / "fastmcp.json"
entrypoint = config.get_entrypoint(config_file)
# Should resolve to absolute path of server file
assert Path(entrypoint.file) == server_file.resolve()
def test_cwd_path_resolution(self, tmp_path):
"""Test that working directory is resolved relative to config."""
import os
# Create directory structure
work_dir = tmp_path / "work"
work_dir.mkdir()
config = FastMCPConfig(
entrypoint="server.py",
deployment={"cwd": "work"}, # type: ignore[arg-type]
)
original_cwd = os.getcwd()
try:
# Apply runtime settings relative to config location
assert config.deployment is not None
config.deployment.apply_runtime_settings(tmp_path / "fastmcp.json")
# Should change to work directory
assert Path.cwd() == work_dir.resolve()
finally:
os.chdir(original_cwd)
def test_requirements_path_resolution(self, tmp_path):
"""Test that requirements path is resolved correctly."""
# Create requirements file
reqs_file = tmp_path / "requirements.txt"
reqs_file.write_text("fastmcp>=2.0")
config = FastMCPConfig(
entrypoint="server.py",
environment={"requirements": "requirements.txt"}, # type: ignore[arg-type]
)
# Build UV args
assert config.environment is not None
uv_args = config.environment.build_uv_args(["fastmcp", "run"])
# Should include requirements file
assert "--with-requirements" in uv_args
req_idx = uv_args.index("--with-requirements") + 1
assert uv_args[req_idx] == "requirements.txt"
class TestConfigValidation:
"""Test configuration validation."""
def test_invalid_transport_rejected(self):
"""Test that invalid transport values are rejected."""
with pytest.raises(ValueError):
FastMCPConfig(
entrypoint="server.py",
deployment={"transport": "invalid_transport"}, # type: ignore[arg-type]
)
def test_streamable_http_transport_rejected(self):
"""Test that streamable-http transport is rejected in fastmcp.json config."""
with pytest.raises(ValueError):
FastMCPConfig(
entrypoint="server.py",
deployment={"transport": "streamable-http"}, # type: ignore[arg-type]
)
def test_invalid_log_level_rejected(self):
"""Test that invalid log level values are rejected."""
with pytest.raises(ValueError):
FastMCPConfig(
entrypoint="server.py",
deployment={"log_level": "INVALID"}, # type: ignore[arg-type]
)
def test_missing_entrypoint_rejected(self):
"""Test that config without entrypoint is rejected."""
with pytest.raises(ValueError):
FastMCPConfig() # type: ignore[call-arg]
def test_valid_transport_values(self):
"""Test that all valid transport values are accepted."""
for transport in ["stdio", "http", "sse"]:
config = FastMCPConfig(
entrypoint="server.py",
deployment={"transport": transport}, # type: ignore[arg-type]
)
assert config.deployment is not None
assert config.deployment.transport == transport
def test_valid_log_levels(self):
"""Test that all valid log levels are accepted."""
for level in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]:
config = FastMCPConfig(
entrypoint="server.py",
deployment={"log_level": level}, # type: ignore[arg-type]
)
assert config.deployment is not None
assert config.deployment.log_level == level
|