mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-21 04:54:17 +02:00
0e5a8fed6c* feat: add Provider abstraction for dynamic components
Introduces a `Provider` base class that allows dynamic provision of
tools, resources, and prompts at runtime. Providers are queried after
static components, enabling database-backed tools, external integrations,
and other dynamic component sources.
```python
from fastmcp import FastMCP, Provider
class DatabaseProvider(Provider):
async def list_tools(self, context):
return await db.fetch_tools()
mcp = FastMCP("Server", providers=[DatabaseProvider()])
```
* Add execution methods to Provider interface
Enable providers to customize how components are executed, not just listed.
This unlocks MountedProvider (future) invoking wrapped server middleware.
- call_tool, read_resource, read_resource_template, render_prompt
- Server uses provider execution methods after filter checks
- Default implementations delegate to component methods
* Apply unified error handling to provider execution
- Document error semantics: list_* gracefully degrades, execution propagates
- Wrap provider call_tool/read_resource/render_prompt with masking logic
- ToolError/ResourceError/PromptError pass through, others wrapped
137 lines
4.2 KiB
Python
137 lines
4.2 KiB
Python
# /// script
|
|
# dependencies = ["aiosqlite", "fastmcp"]
|
|
# ///
|
|
"""
|
|
MCP server with database-configured tools.
|
|
|
|
Tools are loaded from tools.db on each request, so you can add/modify/disable
|
|
tools in the database without restarting the server.
|
|
|
|
Run with: uv run fastmcp run examples/providers/sqlite/server.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import aiosqlite
|
|
from rich import print
|
|
|
|
from fastmcp import Client, FastMCP, Provider
|
|
from fastmcp.server.context import Context
|
|
from fastmcp.tools.tool import Tool, ToolResult
|
|
|
|
DB_PATH = Path(__file__).parent / "tools.db"
|
|
|
|
|
|
class ConfigurableTool(Tool):
|
|
"""A tool that performs a configured arithmetic operation.
|
|
|
|
This demonstrates the pattern: Tool subclass = schema + execution in one place.
|
|
"""
|
|
|
|
operation: str # "add", "multiply", "subtract", "divide"
|
|
default_value: float = 0
|
|
|
|
async def run(self, arguments: dict[str, Any]) -> ToolResult:
|
|
a = arguments.get("a", self.default_value)
|
|
b = arguments.get("b", self.default_value)
|
|
|
|
if self.operation == "add":
|
|
result = a + b
|
|
elif self.operation == "multiply":
|
|
result = a * b
|
|
elif self.operation == "subtract":
|
|
result = a - b
|
|
elif self.operation == "divide":
|
|
if b == 0:
|
|
return ToolResult(
|
|
structured_content={
|
|
"error": "Division by zero",
|
|
"operation": self.operation,
|
|
}
|
|
)
|
|
result = a / b
|
|
else:
|
|
result = a + b
|
|
|
|
return ToolResult(
|
|
structured_content={"result": result, "operation": self.operation}
|
|
)
|
|
|
|
|
|
class SQLiteToolProvider(Provider):
|
|
"""Queries SQLite for tool configurations.
|
|
|
|
Called on every list_tools/get_tool request, so database changes
|
|
are reflected immediately without server restart.
|
|
"""
|
|
|
|
def __init__(self, db_path: str):
|
|
self.db_path = db_path
|
|
|
|
async def list_tools(self, context: Context) -> list[Tool]:
|
|
async with aiosqlite.connect(self.db_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
async with db.execute("SELECT * FROM tools WHERE enabled = 1") as cursor:
|
|
rows = await cursor.fetchall()
|
|
return [self._make_tool(row) for row in rows]
|
|
|
|
async def get_tool(self, context: Context, name: str) -> Tool | None:
|
|
async with aiosqlite.connect(self.db_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
async with db.execute(
|
|
"SELECT * FROM tools WHERE name = ? AND enabled = 1", (name,)
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
return self._make_tool(row) if row else None
|
|
|
|
def _make_tool(self, row: aiosqlite.Row) -> ConfigurableTool:
|
|
return ConfigurableTool(
|
|
name=row["name"],
|
|
description=row["description"],
|
|
parameters=json.loads(row["parameters_schema"]),
|
|
operation=row["operation"],
|
|
default_value=row["default_value"] or 0,
|
|
)
|
|
|
|
|
|
mcp = FastMCP("DynamicToolsServer")
|
|
|
|
provider = SQLiteToolProvider(db_path=str(DB_PATH))
|
|
mcp.add_provider(provider)
|
|
|
|
|
|
@mcp.tool
|
|
def server_info() -> dict[str, str]:
|
|
"""Get information about this server (static tool)."""
|
|
return {
|
|
"name": "DynamicToolsServer",
|
|
"description": "A server with database-configured tools",
|
|
"database": str(DB_PATH),
|
|
}
|
|
|
|
|
|
async def main():
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
print(f"[bold]Available tools ({len(tools)}):[/bold]")
|
|
for tool in tools:
|
|
print(f" • {tool.name}: {tool.description}")
|
|
|
|
print()
|
|
print("[bold]Calling add_numbers(10, 5):[/bold]")
|
|
result = await client.call_tool("add_numbers", {"a": 10, "b": 5})
|
|
print(f" Result: {result.structured_content}")
|
|
|
|
print()
|
|
print("[bold]Calling multiply_numbers(7, 6):[/bold]")
|
|
result = await client.call_tool("multiply_numbers", {"a": 7, "b": 6})
|
|
print(f" Result: {result.structured_content}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|