Files
Pietro Zullo be9942a4d4 feat/python/server (#559)
* feat: server first implementation

* docs: docs for server and images

* chore: locked langchain dependencies in pyproject.toml (#321)

* python/release/1.3.13

* catch uvicorn logs && log session id

* fix debug logic

* docs: logging update

* fix: icon of elicitation

* fix installation paths for local clones

* docs: minor restructuring

* docs: reorganization, killed non existing pages

* docs: beta tag on server

* chore: example servers

* feat: inspector autopopulated and client only

* feat: enhance MCP logging system with customizable formats and improved middleware

* feat: implement weather alert and forecast tools in FastMCP example server

* mcp path configurable

* fix inspector

* export as FastMCP too

* chore: update fastmcp migration example

* lock inspector version

* type runner

* unhide docs and remove dangers

* migration diff

* docs and context example

* export context

* logging improvement

* pretty JSON RPC

* autoConnect url param in inspecotr

* move logging

* type check

* Server Router

* show response logs

* fallback tool name to function name in router registration

* compatibility export

* Introduced ty and prek as type checker and precommit engine respectively - moved ruff config in pyproject toml

* add ty check

* ty runs only on staged files

* ty configuration fix

* move startup info in logging

* remove bug from action

* update claude rules

* fix resource serialization

* comments on inspector errors

* abstracted runner transports

* remove redunant call for network

* fix ip getter

* add inspector mcp path

* response limit logging

* disable ty check in action

* moved transport type to types

* support sse

* show transport type in logs

* change default transport

* refactored tests to use mcp server

* removed fastmcp

* file restructuring

* test table

* sampling, elicitation, tool disabling enabling

* deprecte sse server side

* simplified report

* fix action

* remove sse deprecated test

* move tests files to allow for server integration tests

* docs

* reorg integration test files

* api-reference-doxs

* fix broken links

* telemetry

* remove feature useage from scarf
2025-12-01 14:12:16 +01:00

261 lines
7.6 KiB
Python

"""MCP Router for organizing tools, resources, and prompts into modules."""
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from mcp.types import ToolAnnotations
@dataclass
class PendingTool:
"""A tool waiting to be registered with a server."""
fn: Callable[..., Any]
name: str | None = None
title: str | None = None
description: str | None = None
annotations: ToolAnnotations | None = None
structured_output: bool | None = None
@dataclass
class PendingResource:
"""A resource waiting to be registered with a server."""
fn: Callable[..., Any]
uri: str
name: str | None = None
description: str | None = None
mime_type: str | None = None
@dataclass
class PendingPrompt:
"""A prompt waiting to be registered with a server."""
fn: Callable[..., Any]
name: str | None = None
description: str | None = None
class MCPRouter:
"""
Router for organizing MCP tools, resources, and prompts into reusable modules.
Similar to FastAPI's APIRouter, this allows you to define tools and resources
in separate files and then include them in your main server.
Example:
```python
# routes/math.py
from mcp_use.server import MCPRouter
router = MCPRouter()
@router.tool()
def add(a: int, b: int) -> int:
return a + b
# main.py
from mcp_use.server import MCPServer
from routes.math import router as math_router
server = MCPServer(name="my-server")
server.include_router(math_router, prefix="math")
```
"""
def __init__(
self,
*,
prefix: str = "",
tags: list[str] | None = None,
):
"""
Create a new MCP router.
Args:
prefix: Optional prefix to add to all tool/resource names when included.
For example, prefix="math" would make tool "add" become "math_add".
tags: Optional tags for documentation/organization purposes.
"""
self.prefix = prefix
self.tags = tags or []
self._pending_tools: list[PendingTool] = []
self._pending_resources: list[PendingResource] = []
self._pending_prompts: list[PendingPrompt] = []
def tool(
self,
name: str | None = None,
title: str | None = None,
description: str | None = None,
annotations: ToolAnnotations | None = None,
structured_output: bool | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
"""
Decorator to register a tool with this router.
Args:
name: Override the tool name (defaults to function name)
title: Human-readable title for the tool
description: Tool description (defaults to function docstring)
annotations: MCP tool annotations
structured_output: Whether the tool returns structured output
Example:
```python
@router.tool()
def my_tool(arg: str) -> str:
'''Tool description here.'''
return f"Result: {arg}"
```
"""
def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
self._pending_tools.append(
PendingTool(
fn=fn,
name=name,
title=title,
description=description,
annotations=annotations,
structured_output=structured_output,
)
)
return fn
return decorator
def resource(
self,
uri: str,
name: str | None = None,
description: str | None = None,
mime_type: str | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
"""
Decorator to register a resource with this router.
Args:
uri: Resource URI (required, e.g., "config://app" or "file://data.json")
name: Human-readable name
description: Resource description
mime_type: MIME type of the resource content
Example:
```python
@router.resource(uri="config://app")
def get_config() -> str:
return '{"setting": "value"}'
```
"""
def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
self._pending_resources.append(
PendingResource(
fn=fn,
uri=uri,
name=name,
description=description,
mime_type=mime_type,
)
)
return fn
return decorator
def prompt(
self,
name: str | None = None,
description: str | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
"""
Decorator to register a prompt with this router.
Args:
name: Prompt name (defaults to function name)
description: Prompt description
Example:
```python
@router.prompt()
def greeting_prompt(name: str) -> str:
return f"Hello, {name}! How can I help you today?"
```
"""
def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
self._pending_prompts.append(
PendingPrompt(
fn=fn,
name=name,
description=description,
)
)
return fn
return decorator
def include_router(self, router: "MCPRouter", prefix: str = "") -> None:
"""
Include another router's tools, resources, and prompts into this router.
This allows for nested router organization.
Args:
router: The router to include
prefix: Additional prefix to add to included items
"""
combined_prefix = f"{self.prefix}_{prefix}".strip("_") if prefix else self.prefix
for tool in router._pending_tools:
# Apply combined prefix to tool name
tool_name = tool.name or getattr(tool.fn, "__name__", "unknown")
if combined_prefix:
tool_name = f"{combined_prefix}_{tool_name}"
self._pending_tools.append(
PendingTool(
fn=tool.fn,
name=tool_name,
title=tool.title,
description=tool.description,
annotations=tool.annotations,
structured_output=tool.structured_output,
)
)
for resource in router._pending_resources:
self._pending_resources.append(resource)
for prompt in router._pending_prompts:
prompt_name = prompt.name or getattr(prompt.fn, "__name__", "unknown")
if combined_prefix:
prompt_name = f"{combined_prefix}_{prompt_name}"
self._pending_prompts.append(
PendingPrompt(
fn=prompt.fn,
name=prompt_name,
description=prompt.description,
)
)
@property
def tools(self) -> list[PendingTool]:
"""Get all pending tools in this router."""
return self._pending_tools
@property
def resources(self) -> list[PendingResource]:
"""Get all pending resources in this router."""
return self._pending_resources
@property
def prompts(self) -> list[PendingPrompt]:
"""Get all pending prompts in this router."""
return self._pending_prompts