ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-22 05:24:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #173 from jlowin/prompts

Browse source 
Create MCP prompts on object; fix issue with forwarding proxy templates
This commit is contained in:
Jeremiah Lowin 2025-04-15 12:56:54 -04:00 • committed by GitHub
commit cb049c5052
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 173 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

19
src/fastmcp/prompts/prompt.py
View file

@ -7,6 +7,8 @@ from typing import Annotated, Any, Literal
import pydantic_core
from mcp.types import EmbeddedResource, ImageContent, TextContent
from mcp.types import Prompt as MCPPrompt
from mcp.types import PromptArgument as MCPPromptArgument
from pydantic import BaseModel, BeforeValidator, Field, TypeAdapter, validate_call
from fastmcp.utilities.types import _convert_set_defaults
@ -166,3 +168,20 @@ class Prompt(BaseModel):
if not isinstance(other, Prompt):
return False
return self.model_dump() == other.model_dump()
def to_mcp_prompt(self, **overrides: Any) -> MCPPrompt:
"""Convert the prompt to an MCP prompt."""
arguments = [
MCPPromptArgument(
name=arg.name,
description=arg.description,
required=arg.required,
)
for arg in self.arguments or []
]
kwargs = {
"name": self.name,
"description": self.description,
"arguments": arguments,
}
return MCPPrompt(**kwargs | overrides)

51
src/fastmcp/prompts/prompt_manager.py
View file

@ -1,11 +1,10 @@
"""Prompt management functionality."""
import copy
from collections.abc import Awaitable, Callable
from typing import Any
from fastmcp.exceptions import PromptError
from fastmcp.prompts.prompt import Message, Prompt, PromptResult
from fastmcp.prompts.prompt import MCPPrompt, Message, Prompt, PromptResult
from fastmcp.settings import DuplicateBehavior
from fastmcp.utilities.logging import get_logger
@ -30,13 +29,23 @@ class PromptManager:
self.duplicate_behavior = duplicate_behavior
def get_prompt(self, name: str) -> Prompt | None:
"""Get prompt by name."""
return self._prompts.get(name)
def get_prompt(self, key: str) -> Prompt | None:
"""Get prompt by key."""
return self._prompts.get(key)
def get_prompts(self) -> dict[str, Prompt]:
"""Get all registered prompts, indexed by registered key."""
return self._prompts
def list_prompts(self) -> list[Prompt]:
"""List all registered prompts."""
return list(self._prompts.values())
return list(self.get_prompts().values())
def list_mcp_prompts(self) -> list[MCPPrompt]:
"""List all registered prompts in the format expected by the low-level MCP server."""
return [
prompt.to_mcp_prompt(name=key) for key, prompt in self.get_prompts().items()
]
def add_prompt_from_fn(
self,
@ -49,23 +58,24 @@ class PromptManager:
prompt = Prompt.from_function(fn, name=name, description=description, tags=tags)
return self.add_prompt(prompt)
def add_prompt(self, prompt: Prompt) -> Prompt:
def add_prompt(self, prompt: Prompt, key: str | None = None) -> Prompt:
"""Add a prompt to the manager."""
key = key or prompt.name
# Check for duplicates
existing = self._prompts.get(prompt.name)
existing = self._prompts.get(key)
if existing:
if self.duplicate_behavior == "warn":
logger.warning(f"Prompt already exists: {prompt.name}")
self._prompts[prompt.name] = prompt
logger.warning(f"Prompt already exists: {key}")
self._prompts[key] = prompt
elif self.duplicate_behavior == "replace":
self._prompts[prompt.name] = prompt
self._prompts[key] = prompt
elif self.duplicate_behavior == "error":
raise ValueError(f"Prompt already exists: {prompt.name}")
raise ValueError(f"Prompt already exists: {key}")
elif self.duplicate_behavior == "ignore":
return existing
else:
self._prompts[prompt.name] = prompt
self._prompts[key] = prompt
return prompt
async def render_prompt(
@ -86,19 +96,16 @@ class PromptManager:
Args:
manager: Another PromptManager instance to import prompts from
prefix: Prefix to add to prompt names. The resulting prompt name will
prefix: Prefix to add to prompt names. The resulting prompt key will
be in the format "{prefix}{original_name}" if prefix is provided,
otherwise the original name is used.
For example, with prefix "weather/" and prompt "forecast_prompt",
the imported prompt would be available as "weather/forecast_prompt"
"""
for name, prompt in manager._prompts.items():
# Create prefixed name
prefixed_name = f"{prefix}{name}" if prefix else name
# Create prefixed key
key = f"{prefix}{name}" if prefix else name
new_prompt = copy.copy(prompt)
new_prompt.name = prefixed_name
# Store the prompt with the prefixed name
self.add_prompt(new_prompt)
logger.debug(f'Imported prompt "{name}" as "{prefixed_name}"')
# Store the prompt with the prefixed key
self.add_prompt(prompt, key=key)
logger.debug(f'Imported prompt "{prompt.name}" as "{key}"')

5
src/fastmcp/resources/template.py
View file

@ -6,6 +6,7 @@ import inspect
import re
from collections.abc import Callable
from typing import Annotated, Any
from urllib.parse import unquote
from mcp.types import ResourceTemplate as MCPResourceTemplate
from pydantic import (
@ -38,7 +39,9 @@ def build_regex(template: str) -> re.Pattern:
def match_uri_template(uri: str, uri_template: str) -> dict[str, str] | None:
regex = build_regex(uri_template)
match = regex.match(uri)
return match.groupdict() if match else None
if match:
return {k: unquote(v) for k, v in match.groupdict().items()}
return None
class MyModel(BaseModel):

11
src/fastmcp/server/proxy.py
View file

@ -1,4 +1,5 @@
from typing import Any, cast
from urllib.parse import quote
import mcp.types
from mcp.types import BlobResourceContents, TextResourceContents
@ -104,8 +105,14 @@ class ProxyTemplate(ResourceTemplate):
)
async def create_resource(self, uri: str, params: dict[str, Any]) -> ProxyResource:
# dont use the provided uri, because it may not be the same as the
# uri_template on the remote server.
# quote params to ensure they are valid for the uri_template
parameterized_uri = self.uri_template.format(
**{k: quote(v, safe="") for k, v in params.items()}
)
async with self._client:
result = await self._client.read_resource(uri)
result = await self._client.read_resource(parameterized_uri)
if isinstance(result[0], TextResourceContents):
value = result[0].text
@ -116,7 +123,7 @@ class ProxyTemplate(ResourceTemplate):
return ProxyResource(
client=self._client,
uri=uri,
uri=parameterized_uri,
name=self.name,
description=self.description,
mime_type=result[0].mimeType,

22
src/fastmcp/server/server.py
View file

@ -28,7 +28,6 @@ from mcp.types import (
TextContent,
)
from mcp.types import Prompt as MCPPrompt
from mcp.types import PromptArgument as MCPPromptArgument
from mcp.types import Resource as MCPResource
from mcp.types import ResourceTemplate as MCPResourceTemplate
from mcp.types import Tool as MCPTool
@ -217,11 +216,11 @@ class FastMCP(Generic[LifespanResultT]):
return Context(request_context=request_context, fastmcp=self)
async def call_tool(
self, name: str, arguments: dict[str, Any]
self, key: str, arguments: dict[str, Any]
) -> list[TextContent | ImageContent | EmbeddedResource]:
"""Call a tool by name with arguments."""
context = self.get_context()
result = await self._tool_manager.call_tool(name, arguments, context=context)
result = await self._tool_manager.call_tool(key, arguments, context=context)
converted_result = _convert_to_content(result)
return converted_result
@ -592,22 +591,7 @@ class FastMCP(Generic[LifespanResultT]):
See `list_prompts` for a more ergonomic way to list prompts.
"""
prompts = self.list_prompts()
return [
MCPPrompt(
name=prompt.name,
description=prompt.description,
arguments=[
MCPPromptArgument(
name=arg.name,
description=arg.description,
required=arg.required,
)
for arg in (prompt.arguments or [])
],
)
for prompt in prompts
]
return self._prompt_manager.list_mcp_prompts()
async def get_prompt(
self, name: str, arguments: dict[str, Any] | None = None

12
src/fastmcp/tools/tool_manager.py
View file

@ -36,9 +36,9 @@ class ToolManager:
self.duplicate_behavior = duplicate_behavior
def get_tool(self, name: str) -> Tool | None:
"""Get tool by name."""
return self._tools.get(name)
def get_tool(self, key: str) -> Tool | None:
"""Get tool by key."""
return self._tools.get(key)
def get_tools(self) -> dict[str, Tool]:
"""Get all registered tools, indexed by registered key."""
@ -109,6 +109,6 @@ class ToolManager:
the imported tool would be available as "weather/forecast"
"""
for name, tool in tool_manager._tools.items():
prefixed_name = f"{prefix}{name}" if prefix else name
self.add_tool(tool, key=prefixed_name)
logger.debug(f'Imported tool "{tool.name}" as "{prefixed_name}"')
key = f"{prefix}{name}" if prefix else name
self.add_tool(tool, key=key)
logger.debug(f'Imported tool "{tool.name}" as "{key}"')

21
tests/resources/test_resource_template.py
View file

@ -1,4 +1,5 @@
import json
from urllib.parse import quote
import pytest
from pydantic import BaseModel
@ -312,6 +313,18 @@ class TestMatchUriTemplate:
("test://foo/123", {"x": "foo", "y": "123"}),
("test://bar/456", {"x": "bar", "y": "456"}),
("test://foo/bar", {"x": "foo", "y": "bar"}),
("test://foo/bar/baz", None),
("test://foo/email@domain.com", {"x": "foo", "y": "email@domain.com"}),
("test://two words/foo", {"x": "two words", "y": "foo"}),
("test://two.words/foo+bar", {"x": "two.words", "y": "foo+bar"}),
(
f"test://escaped{quote('/', safe='')}word/bar",
{"x": "escaped/word", "y": "bar"},
),
(
f"test://escaped{quote('{', safe='')}x{quote('}', safe='')}word/bar",
{"x": "escaped{x}word", "y": "bar"},
),
("prefix+test://foo/123", None),
("test://foo", None),
("other://foo/123", None),
@ -361,3 +374,11 @@ class TestMatchUriTemplate:
uri_template = "prefix+test://{x}/test/{y}"
result = match_uri_template(uri=uri, uri_template=uri_template)
assert result == expected_params
def test_quoted_params(self):
uri_template = "user://{name}/{email}"
quoted_name = quote("John Doe", safe="")
quoted_email = quote("john@example.com", safe="")
uri = f"user://{quoted_name}/{quoted_email}"
result = match_uri_template(uri=uri, uri_template=uri_template)
assert result == {"name": "John Doe", "email": "john@example.com"}

84
tests/server/test_mount.py
View file

@ -1,6 +1,7 @@
import contextlib
import json
from urllib.parse import quote
import pytest
from mcp.types import TextContent
from fastmcp.server.server import FastMCP
@ -190,7 +191,6 @@ async def test_mount_multiple_prompts():
assert "sql_explain_sql" in main_app._prompt_manager._prompts
@pytest.mark.anyio
async def test_mount_lifespan():
"""Test that the lifespan of a mounted app is properly handled."""
# Create apps
@ -346,3 +346,83 @@ async def test_mount_with_proxy_tools():
result = await main_app.call_tool("api_get_data", {"query": "test"})
assert isinstance(result[0], TextContent)
assert result[0].text == "Data for query: test"
async def test_mount_with_proxy_prompts():
"""
Test mounting with prompts that have custom keys.
This tests that the prompt's name doesn't change even though the registered
key does, which is important for correct rendering.
"""
# Create apps
main_app = FastMCP("MainApp")
api_app = FastMCP("APIApp")
@api_app.prompt()
def greeting(name: str) -> str:
return f"Hello, {name} from API!"
main_app.mount("api", await FastMCP.as_proxy(api_app))
result = await main_app.get_prompt("api_greeting", {"name": "World"})
assert len(result) > 0
assert isinstance(result[0].content, TextContent)
assert result[0].content.text == "Hello, World from API!"
async def test_mount_with_proxy_resources():
"""
Test mounting with resources that have custom keys.
This tests that the resource's name doesn't change even though the registered
key does, which is important for correct access.
"""
# Create apps
main_app = FastMCP("MainApp")
api_app = FastMCP("APIApp")
# Create a resource in the API app
@api_app.resource(uri="config://settings")
def get_config():
return {
"api_key": "12345",
"base_url": "https://api.example.com",
}
main_app.mount("api", await FastMCP.as_proxy(api_app))
# Access the resource through the main app with the prefixed key
resource = await main_app.read_resource("api+config://settings")
assert resource is not None
resource = json.loads(resource)
assert resource["api_key"] == "12345"
assert resource["base_url"] == "https://api.example.com"
async def test_mount_with_proxy_resource_templates():
"""
Test mounting with resource templates that have custom keys.
This tests that the template's name doesn't change even though the registered
key does, which is important for correct instantiation.
"""
# Create apps
main_app = FastMCP("MainApp")
api_app = FastMCP("APIApp")
# Create a resource template in the API app
@api_app.resource(uri="user://{name}/{email}")
def create_user(name: str, email: str):
return {"name": name, "email": email}
main_app.mount("api", await FastMCP.as_proxy(api_app))
# Instantiate the template through the main app with the prefixed key
quoted_name = quote("John Doe", safe="")
quoted_email = quote("john@example.com", safe="")
user = await main_app.read_resource(f"api+user://{quoted_name}/{quoted_email}")
assert user is not None
user = json.loads(user)
assert user["name"] == "John Doe"
assert user["email"] == "john@example.com"