mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-18 11:39:12 +02:00
748 lines
25 KiB
Python
748 lines
25 KiB
Python
"""Tests for Docket-style dependency injection in FastMCP."""
|
|
|
|
from contextlib import asynccontextmanager, contextmanager
|
|
|
|
import mcp.types as mcp_types
|
|
import pytest
|
|
from mcp.types import TextContent, TextResourceContents
|
|
|
|
from fastmcp import FastMCP
|
|
from fastmcp.client import Client
|
|
from fastmcp.dependencies import CurrentContext, Depends
|
|
from fastmcp.server.context import Context
|
|
|
|
HUZZAH = "huzzah!"
|
|
|
|
|
|
class Connection:
|
|
"""Test connection that tracks whether it's currently open."""
|
|
|
|
def __init__(self):
|
|
self.is_open = False
|
|
|
|
async def __aenter__(self):
|
|
self.is_open = True
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
|
self.is_open = False
|
|
|
|
|
|
@asynccontextmanager
|
|
async def get_connection():
|
|
"""Dependency that provides an open connection."""
|
|
async with Connection() as conn:
|
|
yield conn
|
|
|
|
|
|
@pytest.fixture
|
|
def mcp():
|
|
"""Create a FastMCP server for testing."""
|
|
return FastMCP("test-server")
|
|
|
|
|
|
async def test_depends_with_sync_function(mcp: FastMCP):
|
|
"""Test that Depends works with sync dependency functions."""
|
|
|
|
def get_config() -> dict[str, str]:
|
|
return {"api_key": "secret123", "endpoint": "https://api.example.com"}
|
|
|
|
@mcp.tool()
|
|
def fetch_data(query: str, config: dict[str, str] = Depends(get_config)) -> str:
|
|
return (
|
|
f"Fetching '{query}' from {config['endpoint']} with key {config['api_key']}"
|
|
)
|
|
|
|
result = await mcp.call_tool("fetch_data", {"query": "users"})
|
|
assert result.structured_content is not None
|
|
text = result.structured_content["result"]
|
|
assert "Fetching 'users' from https://api.example.com" in text
|
|
assert "secret123" in text
|
|
|
|
|
|
async def test_depends_with_async_function(mcp: FastMCP):
|
|
"""Test that Depends works with async dependency functions."""
|
|
|
|
async def get_user_id() -> int:
|
|
return 42
|
|
|
|
@mcp.tool()
|
|
async def greet_user(name: str, user_id: int = Depends(get_user_id)) -> str: # type: ignore[assignment]
|
|
return f"Hello {name}, your ID is {user_id}"
|
|
|
|
result = await mcp.call_tool("greet_user", {"name": "Alice"})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "Hello Alice, your ID is 42"
|
|
|
|
|
|
async def test_depends_with_async_context_manager(mcp: FastMCP):
|
|
"""Test that Depends works with async context managers for resource management."""
|
|
cleanup_called = False
|
|
|
|
@asynccontextmanager
|
|
async def get_database():
|
|
db = "db_connection"
|
|
try:
|
|
yield db
|
|
finally:
|
|
nonlocal cleanup_called
|
|
cleanup_called = True
|
|
|
|
@mcp.tool()
|
|
async def query_db(sql: str, db: str = Depends(get_database)) -> str: # type: ignore[assignment]
|
|
return f"Executing '{sql}' on {db}"
|
|
|
|
result = await mcp.call_tool("query_db", {"sql": "SELECT * FROM users"})
|
|
assert result.structured_content is not None
|
|
assert (
|
|
"Executing 'SELECT * FROM users' on db_connection"
|
|
in result.structured_content["result"]
|
|
)
|
|
assert cleanup_called
|
|
|
|
|
|
async def test_nested_dependencies(mcp: FastMCP):
|
|
"""Test that dependencies can depend on other dependencies."""
|
|
|
|
def get_base_url() -> str:
|
|
return "https://api.example.com"
|
|
|
|
def get_api_client(base_url: str = Depends(get_base_url)) -> dict[str, str]:
|
|
return {"base_url": base_url, "version": "v1"}
|
|
|
|
@mcp.tool()
|
|
async def call_api(
|
|
endpoint: str, client: dict[str, str] = Depends(get_api_client)
|
|
) -> str:
|
|
return f"Calling {client['base_url']}/{client['version']}/{endpoint}"
|
|
|
|
result = await mcp.call_tool("call_api", {"endpoint": "users"})
|
|
assert result.structured_content is not None
|
|
assert (
|
|
result.structured_content["result"]
|
|
== "Calling https://api.example.com/v1/users"
|
|
)
|
|
|
|
|
|
async def test_dependencies_excluded_from_schema(mcp: FastMCP):
|
|
"""Test that dependency parameters don't appear in the tool schema."""
|
|
|
|
def get_config() -> dict[str, str]:
|
|
return {"key": "value"}
|
|
|
|
@mcp.tool()
|
|
async def my_tool(
|
|
name: str, age: int, config: dict[str, str] = Depends(get_config)
|
|
) -> str:
|
|
return f"{name} is {age} years old"
|
|
|
|
result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
|
|
tool = next(t for t in result.tools if t.name == "my_tool")
|
|
|
|
assert "name" in tool.inputSchema["properties"]
|
|
assert "age" in tool.inputSchema["properties"]
|
|
assert "config" not in tool.inputSchema["properties"]
|
|
assert len(tool.inputSchema["properties"]) == 2
|
|
|
|
|
|
async def test_current_context_dependency(mcp: FastMCP):
|
|
"""Test that CurrentContext dependency provides access to FastMCP Context."""
|
|
|
|
@mcp.tool()
|
|
def use_context(ctx: Context = CurrentContext()) -> str:
|
|
assert isinstance(ctx, Context)
|
|
return HUZZAH
|
|
|
|
result = await mcp.call_tool("use_context", {})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == HUZZAH
|
|
|
|
|
|
async def test_current_context_and_legacy_context_coexist(mcp: FastMCP):
|
|
"""Test that CurrentContext dependency and legacy Context injection work together."""
|
|
|
|
@mcp.tool()
|
|
def use_both_contexts(
|
|
legacy_ctx: Context,
|
|
dep_ctx: Context = CurrentContext(),
|
|
) -> str:
|
|
assert isinstance(legacy_ctx, Context)
|
|
assert isinstance(dep_ctx, Context)
|
|
assert legacy_ctx is dep_ctx
|
|
return HUZZAH
|
|
|
|
result = await mcp.call_tool("use_both_contexts", {})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == HUZZAH
|
|
|
|
|
|
async def test_backward_compat_context_still_works(mcp: FastMCP):
|
|
"""Test that existing Context injection via type annotation still works."""
|
|
|
|
@mcp.tool()
|
|
async def get_request_id(ctx: Context) -> str:
|
|
return ctx.request_id
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("get_request_id", {})
|
|
assert len(result.content) == 1
|
|
content = result.content[0]
|
|
assert isinstance(content, TextContent)
|
|
assert len(content.text) > 0
|
|
|
|
|
|
async def test_sync_tool_with_async_dependency(mcp: FastMCP):
|
|
"""Test that sync tools work with async dependencies."""
|
|
|
|
async def fetch_config() -> str:
|
|
return "loaded_config"
|
|
|
|
@mcp.tool()
|
|
def process_data(value: int, config: str = Depends(fetch_config)) -> str: # type: ignore[assignment]
|
|
return f"Processing {value} with {config}"
|
|
|
|
result = await mcp.call_tool("process_data", {"value": 100})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "Processing 100 with loaded_config"
|
|
|
|
|
|
async def test_dependency_caching(mcp: FastMCP):
|
|
"""Test that dependencies are cached within a single tool call."""
|
|
call_count = 0
|
|
|
|
def expensive_dependency() -> int:
|
|
nonlocal call_count
|
|
call_count += 1
|
|
return 42
|
|
|
|
@mcp.tool()
|
|
async def tool_with_cached_dep(
|
|
dep1: int = Depends(expensive_dependency),
|
|
dep2: int = Depends(expensive_dependency),
|
|
) -> str:
|
|
return f"{dep1} + {dep2} = {dep1 + dep2}"
|
|
|
|
result = await mcp.call_tool("tool_with_cached_dep", {})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "42 + 42 = 84"
|
|
assert call_count == 1
|
|
|
|
|
|
async def test_context_and_depends_together(mcp: FastMCP):
|
|
"""Test that Context type injection and Depends can be used together."""
|
|
|
|
def get_multiplier() -> int:
|
|
return 10
|
|
|
|
@mcp.tool()
|
|
async def mixed_deps(
|
|
value: int, ctx: Context, multiplier: int = Depends(get_multiplier)
|
|
) -> str:
|
|
assert isinstance(ctx, Context)
|
|
assert ctx.request_id
|
|
assert len(ctx.request_id) > 0
|
|
return (
|
|
f"Request {ctx.request_id}: {value} * {multiplier} = {value * multiplier}"
|
|
)
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("mixed_deps", {"value": 5})
|
|
assert len(result.content) == 1
|
|
content = result.content[0]
|
|
assert isinstance(content, TextContent)
|
|
assert "5 * 10 = 50" in content.text
|
|
assert "Request " in content.text
|
|
|
|
|
|
async def test_resource_with_dependency(mcp: FastMCP):
|
|
"""Test that resources support dependency injection."""
|
|
|
|
def get_storage_path() -> str:
|
|
return "/data/config"
|
|
|
|
@mcp.resource("config://settings")
|
|
async def get_settings(storage: str = Depends(get_storage_path)) -> str:
|
|
return f"Settings loaded from {storage}"
|
|
|
|
result = await mcp.read_resource("config://settings")
|
|
assert len(result.contents) == 1
|
|
assert result.contents[0].content == "Settings loaded from /data/config"
|
|
|
|
|
|
async def test_resource_with_context_and_dependency(mcp: FastMCP):
|
|
"""Test that resources can use both Context and Depends."""
|
|
|
|
def get_prefix() -> str:
|
|
return "DATA"
|
|
|
|
@mcp.resource("config://info")
|
|
async def get_info(ctx: Context, prefix: str = Depends(get_prefix)) -> str:
|
|
return f"{prefix}: Request {ctx.request_id}"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("config://info")
|
|
assert len(result) == 1
|
|
content = result[0]
|
|
assert isinstance(content, TextResourceContents)
|
|
assert "DATA: Request " in content.text
|
|
assert len(content.text.split("Request ")[1]) > 0
|
|
|
|
|
|
async def test_prompt_with_dependency(mcp: FastMCP):
|
|
"""Test that prompts support dependency injection."""
|
|
|
|
def get_tone() -> str:
|
|
return "friendly and helpful"
|
|
|
|
@mcp.prompt()
|
|
async def custom_prompt(topic: str, tone: str = Depends(get_tone)) -> str:
|
|
return f"Write about {topic} in a {tone} tone"
|
|
|
|
result = await mcp.render_prompt("custom_prompt", {"topic": "Python"})
|
|
assert len(result.messages) == 1
|
|
message = result.messages[0]
|
|
content = message.content
|
|
assert isinstance(content, TextContent)
|
|
assert content.text == "Write about Python in a friendly and helpful tone"
|
|
|
|
|
|
async def test_prompt_with_context_and_dependency(mcp: FastMCP):
|
|
"""Test that prompts can use both Context and Depends."""
|
|
|
|
def get_style() -> str:
|
|
return "concise"
|
|
|
|
@mcp.prompt()
|
|
async def styled_prompt(
|
|
query: str, ctx: Context, style: str = Depends(get_style)
|
|
) -> str:
|
|
assert isinstance(ctx, Context)
|
|
assert ctx.request_id
|
|
return f"Answer '{query}' in a {style} style"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.get_prompt("styled_prompt", {"query": "What is MCP?"})
|
|
assert len(result.messages) == 1
|
|
message = result.messages[0]
|
|
content = message.content
|
|
assert isinstance(content, TextContent)
|
|
assert content.text == "Answer 'What is MCP?' in a concise style"
|
|
|
|
|
|
async def test_resource_template_with_dependency(mcp: FastMCP):
|
|
"""Test that resource templates support dependency injection."""
|
|
|
|
def get_base_path() -> str:
|
|
return "/var/data"
|
|
|
|
@mcp.resource("data://{filename}")
|
|
async def get_file(filename: str, base_path: str = Depends(get_base_path)) -> str:
|
|
return f"Reading {base_path}/{filename}"
|
|
|
|
result = await mcp.read_resource("data://config.txt")
|
|
assert len(result.contents) == 1
|
|
assert result.contents[0].content == "Reading /var/data/config.txt"
|
|
|
|
|
|
async def test_resource_template_with_context_and_dependency(mcp: FastMCP):
|
|
"""Test that resource templates can use both Context and Depends."""
|
|
|
|
def get_version() -> str:
|
|
return "v2"
|
|
|
|
@mcp.resource("api://{endpoint}")
|
|
async def call_endpoint(
|
|
endpoint: str, ctx: Context, version: str = Depends(get_version)
|
|
) -> str:
|
|
assert isinstance(ctx, Context)
|
|
assert ctx.request_id
|
|
return f"Calling {version}/{endpoint}"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("api://users")
|
|
assert len(result) == 1
|
|
content = result[0]
|
|
assert isinstance(content, TextResourceContents)
|
|
assert content.text == "Calling v2/users"
|
|
|
|
|
|
async def test_async_tool_context_manager_stays_open(mcp: FastMCP):
|
|
"""Test that context manager dependencies stay open during async tool execution.
|
|
|
|
Context managers must remain open while the async function executes, not just
|
|
while it's being called (which only returns a coroutine).
|
|
"""
|
|
|
|
@mcp.tool()
|
|
async def query_data(
|
|
query: str,
|
|
connection: Connection = Depends(get_connection), # type: ignore[assignment]
|
|
) -> str:
|
|
assert connection.is_open
|
|
return f"open={connection.is_open}"
|
|
|
|
result = await mcp.call_tool("query_data", {"query": "test"})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "open=True"
|
|
|
|
|
|
async def test_async_resource_context_manager_stays_open(mcp: FastMCP):
|
|
"""Test that context manager dependencies stay open during async resource execution."""
|
|
|
|
@mcp.resource("data://config")
|
|
async def load_config(connection: Connection = Depends(get_connection)) -> str: # type: ignore[assignment]
|
|
assert connection.is_open
|
|
return f"open={connection.is_open}"
|
|
|
|
result = await mcp.read_resource("data://config")
|
|
assert result.contents[0].content == "open=True"
|
|
|
|
|
|
async def test_async_resource_template_context_manager_stays_open(mcp: FastMCP):
|
|
"""Test that context manager dependencies stay open during async resource template execution."""
|
|
|
|
@mcp.resource("user://{user_id}")
|
|
async def get_user(
|
|
user_id: str,
|
|
connection: Connection = Depends(get_connection), # type: ignore[assignment]
|
|
) -> str:
|
|
assert connection.is_open
|
|
return f"open={connection.is_open},user={user_id}"
|
|
|
|
result = await mcp.read_resource("user://123")
|
|
assert isinstance(result.contents[0].content, str)
|
|
assert "open=True" in result.contents[0].content
|
|
|
|
|
|
async def test_async_prompt_context_manager_stays_open(mcp: FastMCP):
|
|
"""Test that context manager dependencies stay open during async prompt execution."""
|
|
|
|
@mcp.prompt()
|
|
async def research_prompt(
|
|
topic: str,
|
|
connection: Connection = Depends(get_connection), # type: ignore[assignment]
|
|
) -> str:
|
|
assert connection.is_open
|
|
return f"open={connection.is_open},topic={topic}"
|
|
|
|
result = await mcp.render_prompt("research_prompt", {"topic": "AI"})
|
|
message = result.messages[0]
|
|
content = message.content
|
|
assert isinstance(content, TextContent)
|
|
assert "open=True" in content.text
|
|
|
|
|
|
async def test_argument_validation_with_dependencies(mcp: FastMCP):
|
|
"""Test that user arguments are still validated when dependencies are present."""
|
|
|
|
def get_config() -> dict[str, str]:
|
|
return {"key": "value"}
|
|
|
|
@mcp.tool()
|
|
async def validated_tool(
|
|
age: int, # Should validate type
|
|
config: dict[str, str] = Depends(get_config),
|
|
) -> str:
|
|
return f"age={age}"
|
|
|
|
# Valid argument
|
|
result = await mcp.call_tool("validated_tool", {"age": 25})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "age=25"
|
|
|
|
# Invalid argument type should fail validation
|
|
import pydantic
|
|
|
|
with pytest.raises(pydantic.ValidationError):
|
|
await mcp.call_tool("validated_tool", {"age": "not a number"})
|
|
|
|
|
|
async def test_connection_dependency_excluded_from_tool_schema(mcp: FastMCP):
|
|
"""Test that Connection dependency parameter is excluded from tool schema."""
|
|
|
|
@mcp.tool()
|
|
async def with_connection(
|
|
name: str,
|
|
connection: Connection = Depends(get_connection), # type: ignore[assignment]
|
|
) -> str:
|
|
return name
|
|
|
|
result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
|
|
tool = next(t for t in result.tools if t.name == "with_connection")
|
|
|
|
assert "name" in tool.inputSchema["properties"]
|
|
assert "connection" not in tool.inputSchema["properties"]
|
|
|
|
|
|
async def test_sync_tool_context_manager_stays_open(mcp: FastMCP):
|
|
"""Test that sync context manager dependencies work with tools."""
|
|
conn = Connection()
|
|
|
|
@contextmanager
|
|
def get_sync_connection():
|
|
conn.is_open = True
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.is_open = False
|
|
|
|
@mcp.tool()
|
|
async def query_sync(
|
|
query: str,
|
|
connection: Connection = Depends(get_sync_connection), # type: ignore[assignment]
|
|
) -> str:
|
|
assert connection.is_open
|
|
return f"open={connection.is_open}"
|
|
|
|
result = await mcp.call_tool("query_sync", {"query": "test"})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == "open=True"
|
|
assert not conn.is_open
|
|
|
|
|
|
async def test_sync_resource_context_manager_stays_open(mcp: FastMCP):
|
|
"""Test that sync context manager dependencies work with resources."""
|
|
conn = Connection()
|
|
|
|
@contextmanager
|
|
def get_sync_connection():
|
|
conn.is_open = True
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.is_open = False
|
|
|
|
@mcp.resource("data://sync")
|
|
async def load_sync(connection: Connection = Depends(get_sync_connection)) -> str: # type: ignore[assignment]
|
|
assert connection.is_open
|
|
return f"open={connection.is_open}"
|
|
|
|
result = await mcp.read_resource("data://sync")
|
|
assert result.contents[0].content == "open=True"
|
|
assert not conn.is_open
|
|
|
|
|
|
async def test_sync_resource_template_context_manager_stays_open(mcp: FastMCP):
|
|
"""Test that sync context manager dependencies work with resource templates."""
|
|
conn = Connection()
|
|
|
|
@contextmanager
|
|
def get_sync_connection():
|
|
conn.is_open = True
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.is_open = False
|
|
|
|
@mcp.resource("item://{item_id}")
|
|
async def get_item(
|
|
item_id: str,
|
|
connection: Connection = Depends(get_sync_connection), # type: ignore[assignment]
|
|
) -> str:
|
|
assert connection.is_open
|
|
return f"open={connection.is_open},item={item_id}"
|
|
|
|
result = await mcp.read_resource("item://456")
|
|
assert isinstance(result.contents[0].content, str)
|
|
assert "open=True" in result.contents[0].content
|
|
assert not conn.is_open
|
|
|
|
|
|
async def test_sync_prompt_context_manager_stays_open(mcp: FastMCP):
|
|
"""Test that sync context manager dependencies work with prompts."""
|
|
conn = Connection()
|
|
|
|
@contextmanager
|
|
def get_sync_connection():
|
|
conn.is_open = True
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.is_open = False
|
|
|
|
@mcp.prompt()
|
|
async def sync_prompt(
|
|
topic: str,
|
|
connection: Connection = Depends(get_sync_connection), # type: ignore[assignment]
|
|
) -> str:
|
|
assert connection.is_open
|
|
return f"open={connection.is_open},topic={topic}"
|
|
|
|
result = await mcp.render_prompt("sync_prompt", {"topic": "test"})
|
|
message = result.messages[0]
|
|
content = message.content
|
|
assert isinstance(content, TextContent)
|
|
assert "open=True" in content.text
|
|
assert not conn.is_open
|
|
|
|
|
|
async def test_external_user_cannot_override_dependency(mcp: FastMCP):
|
|
"""Test that external MCP clients cannot override dependency parameters."""
|
|
|
|
def get_admin_status() -> str:
|
|
return "not_admin"
|
|
|
|
@mcp.tool()
|
|
async def check_permission(
|
|
action: str, admin: str = Depends(get_admin_status)
|
|
) -> str:
|
|
return f"action={action},admin={admin}"
|
|
|
|
# Verify dependency is NOT in the schema
|
|
result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
|
|
tool = next(t for t in result.tools if t.name == "check_permission")
|
|
assert "admin" not in tool.inputSchema["properties"]
|
|
|
|
# Normal call - dependency is resolved
|
|
result = await mcp.call_tool("check_permission", {"action": "read"})
|
|
assert result.structured_content is not None
|
|
assert "admin=not_admin" in result.structured_content["result"]
|
|
|
|
# Try to override dependency - rejected (not in schema)
|
|
import pydantic
|
|
|
|
with pytest.raises(pydantic.ValidationError):
|
|
await mcp.call_tool("check_permission", {"action": "read", "admin": "hacker"})
|
|
|
|
|
|
async def test_prompt_dependency_cannot_be_overridden_externally(mcp: FastMCP):
|
|
"""Test that external callers cannot override prompt dependencies.
|
|
|
|
This is a security test - dependencies should NEVER be overridable from
|
|
outside the server, even for prompts which don't validate against strict schemas.
|
|
"""
|
|
|
|
def get_secret() -> str:
|
|
return "real_secret"
|
|
|
|
@mcp.prompt()
|
|
async def secure_prompt(topic: str, secret: str = Depends(get_secret)) -> str:
|
|
return f"Topic: {topic}, Secret: {secret}"
|
|
|
|
# Normal call - should use dependency
|
|
result = await mcp.render_prompt("secure_prompt", {"topic": "test"})
|
|
message = result.messages[0]
|
|
content = message.content
|
|
assert isinstance(content, TextContent)
|
|
assert "Secret: real_secret" in content.text
|
|
|
|
# Try to override dependency - should be ignored/rejected
|
|
result = await mcp.render_prompt(
|
|
"secure_prompt",
|
|
{"topic": "test", "secret": "HACKED"}, # Attempt override
|
|
)
|
|
message = result.messages[0]
|
|
content = message.content
|
|
assert isinstance(content, TextContent)
|
|
# Should still use real dependency, not hacked value
|
|
assert "Secret: real_secret" in content.text
|
|
assert "HACKED" not in content.text
|
|
|
|
|
|
async def test_resource_dependency_cannot_be_overridden_externally(mcp: FastMCP):
|
|
"""Test that external callers cannot override resource dependencies."""
|
|
|
|
def get_api_key() -> str:
|
|
return "real_api_key"
|
|
|
|
@mcp.resource("data://config")
|
|
async def get_config(api_key: str = Depends(get_api_key)) -> str:
|
|
return f"API Key: {api_key}"
|
|
|
|
# Normal call
|
|
result = await mcp.read_resource("data://config")
|
|
assert isinstance(result.contents[0].content, str)
|
|
assert "API Key: real_api_key" in result.contents[0].content
|
|
|
|
# Resources don't accept arguments from clients (static URI)
|
|
# so this scenario is less of a concern, but documenting it
|
|
|
|
|
|
async def test_resource_template_dependency_cannot_be_overridden_externally(
|
|
mcp: FastMCP,
|
|
):
|
|
"""Test that external callers cannot override resource template dependencies.
|
|
|
|
Resource templates extract parameters from the URI path, so there's a risk
|
|
that a dependency parameter name could match a URI parameter.
|
|
"""
|
|
|
|
def get_auth_token() -> str:
|
|
return "real_token"
|
|
|
|
@mcp.resource("user://{user_id}")
|
|
async def get_user(user_id: str, token: str = Depends(get_auth_token)) -> str:
|
|
return f"User: {user_id}, Token: {token}"
|
|
|
|
# Normal call
|
|
result = await mcp.read_resource("user://123")
|
|
assert isinstance(result.contents[0].content, str)
|
|
assert "User: 123, Token: real_token" in result.contents[0].content
|
|
|
|
# Try to inject token via URI (shouldn't be possible with this pattern)
|
|
# But if URI was user://{token}, it could extract it
|
|
|
|
|
|
async def test_resource_template_uri_cannot_match_dependency_name(mcp: FastMCP):
|
|
"""Test that URI parameters cannot have the same name as dependencies.
|
|
|
|
If a URI template tries to use a parameter name that's also a dependency,
|
|
the template creation should fail because the dependency is excluded from
|
|
the user-facing signature.
|
|
"""
|
|
|
|
def get_token() -> str:
|
|
return "real_token"
|
|
|
|
# This should fail - {token} in URI but token is a dependency parameter
|
|
with pytest.raises(ValueError, match="URI parameters.*must be a subset"):
|
|
|
|
@mcp.resource("auth://{token}/validate")
|
|
async def validate(token: str = Depends(get_token)) -> str:
|
|
return f"Validating with: {token}"
|
|
|
|
|
|
async def test_toolerror_propagates_from_dependency(mcp: FastMCP):
|
|
"""ToolError raised in a dependency should propagate unchanged (issue #2633).
|
|
|
|
When a dependency raises ToolError, it should not be wrapped in RuntimeError.
|
|
This allows developers to use ToolError for validation in dependencies.
|
|
"""
|
|
from fastmcp.exceptions import ToolError
|
|
|
|
def validate_client_id() -> str:
|
|
raise ToolError("Client ID is required - select a client first")
|
|
|
|
@mcp.tool()
|
|
async def my_tool(client_id: str = Depends(validate_client_id)) -> str:
|
|
return f"Working with client: {client_id}"
|
|
|
|
async with Client(mcp) as client:
|
|
# ToolError is converted to an error result by the server
|
|
result = await client.call_tool("my_tool", {}, raise_on_error=False)
|
|
assert result.is_error
|
|
# The original error message should be preserved (not wrapped in RuntimeError)
|
|
assert isinstance(result.content[0], TextContent)
|
|
assert result.content[0].text == "Client ID is required - select a client first"
|
|
|
|
|
|
async def test_validation_error_propagates_from_dependency(mcp: FastMCP):
|
|
"""ValidationError raised in a dependency should propagate unchanged."""
|
|
from fastmcp.exceptions import ValidationError
|
|
|
|
def validate_input() -> str:
|
|
raise ValidationError("Invalid input format")
|
|
|
|
@mcp.tool()
|
|
async def tool_with_validation(val: str = Depends(validate_input)) -> str:
|
|
return val
|
|
|
|
async with Client(mcp) as client:
|
|
# ValidationError is re-raised by the server and becomes an error result
|
|
# The original error message should be preserved (not wrapped in RuntimeError)
|
|
result = await client.call_tool(
|
|
"tool_with_validation", {}, raise_on_error=False
|
|
)
|
|
assert result.is_error
|
|
assert isinstance(result.content[0], TextContent)
|
|
assert result.content[0].text == "Invalid input format"
|