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

Forward guard asks through prompt, resource, and template proxies

Browse source 
A proxy has no back-channel to the real user, so driving a backend ask inside
it failed outright. Surface it as a result for the parent, as ProxyTool does.
This commit is contained in:
Jeremiah Lowin 2026-07-26 16:58:58 -04:00
commit f7eed91aa8
No known key found for this signature in database
3 changed files with 313 additions and 138 deletions
Download patch file Download diff file Expand all files Collapse all files

89
fastmcp_slim/fastmcp/server/providers/proxy.py
View file

@ -42,9 +42,13 @@ from fastmcp.client.transports.base import TransportOptions
from fastmcp.exceptions import ResourceError
from fastmcp.mcp_config import MCPConfig
from fastmcp.prompts import Message, Prompt, PromptResult
from fastmcp.prompts.base import PromptArgument
from fastmcp.prompts.base import InputRequiredPromptResult, PromptArgument
from fastmcp.resources import Resource, ResourceTemplate
from fastmcp.resources.base import ResourceContent, ResourceResult
from fastmcp.resources.base import (
InputRequiredResourceResult,
ResourceContent,
ResourceResult,
)
from fastmcp.resources.template import expand_uri_template
from fastmcp.server.context import Context
from fastmcp.server.dependencies import fastmcp_request_ctx, get_context
@ -118,6 +122,36 @@ def _proxy_upstream_error(error: Exception) -> MCPError:
)
async def _relay_read_resource(
client: Client, uri: str, ctx: Context | None
) -> (
list[mcp_types.TextResourceContents | mcp_types.BlobResourceContents]
| mcp_types.InputRequiredResult
):
"""Read a backend resource, surfacing a guard ask rather than driving it.
Mirrors `ProxyTool.run`: on a modern backend the low-level session is used
so an `InputRequiredResult` (SEP-2322) comes back as a result for the parent
to forward, instead of the high-level client trying to answer it here — the
proxy has no back-channel to the real user, so driving it fails outright.
The inbound request's continuation state travels down so the backend guard
sees the client's answers on its own `ctx.input_responses`.
"""
if client.protocol_version not in MODERN_PROTOCOL_VERSIONS:
return await client.read_resource(uri)
result = await client._await_with_session_monitoring(
client.session.read_resource(
uri,
input_responses=ctx.input_responses if ctx else None,
request_state=ctx.request_state if ctx else None,
allow_input_required=True,
)
)
if isinstance(result, mcp_types.InputRequiredResult):
return result
return list(result.contents)
def _stash_proxy_request_context(client: Client, ctx: Context) -> None:
"""Stash the proxy's ``RequestContext`` on a ``ProxyClient`` before a backend call.
@ -418,9 +452,12 @@ class ProxyResource(Resource):
) as span:
span.set_attribute("fastmcp.provider.type", "ProxyProvider")
client = await self._get_client()
ctx = get_context()
async with client:
_stash_proxy_request_context(client, get_context())
result = await client.read_resource(backend_uri)
_stash_proxy_request_context(client, ctx)
result = await _relay_read_resource(client, backend_uri, ctx)
if isinstance(result, mcp_types.InputRequiredResult):
return InputRequiredResourceResult(result)
if not result:
raise ResourceError(
f"Remote server returned empty content for {backend_uri}"
@ -516,9 +553,28 @@ class ProxyTemplate(ResourceTemplate):
backend_template = self._backend_uri_template or self.uri_template
parameterized_uri = expand_uri_template(backend_template, params)
client = await self._get_client()
ctx = context or get_context()
async with client:
_stash_proxy_request_context(client, context or get_context())
result = await client.read_resource(parameterized_uri)
_stash_proxy_request_context(client, ctx)
result = await _relay_read_resource(client, parameterized_uri, ctx)
if isinstance(result, mcp_types.InputRequiredResult):
# The backend template asked for input. `InputRequiredResourceResult`
# is a `ResourceResult`, so caching it on the returned resource lets
# the ask ride the ordinary read path out to the parent's wire
# handler, which unwraps it.
return ProxyResource(
client_factory=self._client_factory,
uri=parameterized_uri,
name=self.name,
title=self.title,
description=self.description,
mime_type=self.mime_type or "text/plain",
icons=self.icons,
meta=self.meta,
tags=get_fastmcp_metadata(self.meta).get("tags", []),
_cached_content=InputRequiredResourceResult(result),
)
if not result:
raise ResourceError(
@ -635,9 +691,26 @@ class ProxyPrompt(Prompt):
) as span:
span.set_attribute("fastmcp.provider.type", "ProxyProvider")
client = await self._get_client()
ctx = get_context()
async with client:
_stash_proxy_request_context(client, get_context())
result = await client.get_prompt(backend_name, arguments)
_stash_proxy_request_context(client, ctx)
if client.protocol_version in MODERN_PROTOCOL_VERSIONS:
# See `_relay_read_resource`: surface a backend guard's ask
# instead of trying to answer it inside the proxy.
raw = await client._await_with_session_monitoring(
client.session.get_prompt(
backend_name,
arguments,
input_responses=ctx.input_responses if ctx else None,
request_state=ctx.request_state if ctx else None,
allow_input_required=True,
)
)
if isinstance(raw, mcp_types.InputRequiredResult):
return InputRequiredPromptResult(raw)
result = raw
else:
result = await client.get_prompt(backend_name, arguments)
# Convert GetPromptResult to PromptResult, preserving meta from result
# (not the static prompt meta which includes fastmcp tags)
# Convert PromptMessages to Messages

130
tests/server/test_mrtr_guards.py
View file

@ -1215,133 +1215,3 @@ class TestHttpTransport:
assert asked == ["Where would you like to fly?", "When to Paris?"]
assert result.data == "Booked Paris on 2026-08-01"
class TestPromptGuard:
"""`InputRequiredResult` is a result type, not a tools/call feature, so a
prompt can ask for input the same way a tool does (SEP-2322)."""
@staticmethod
def _context_prompt_server() -> FastMCP:
mcp = FastMCP("prompt-guard")
@mcp.prompt
async def summarize(ctx: Context) -> str | InputRequiredResult:
responses = ctx.input_responses
if responses is None:
return _ask(
_elicit("context", "What context?", "context"),
key="context",
request_state=None,
)
return f"Summarizing with {_accepted(responses, 'context')['context']}"
return mcp
async def test_prompt_emits_input_required(self):
"""The asking round reaches the wire as an InputRequiredResult."""
async with Client(self._context_prompt_server(), mode="auto") as client:
result = await client.session.get_prompt(
"summarize", allow_input_required=True
)
assert isinstance(result, InputRequiredResult)
assert "context" in result.input_requests
async def test_prompt_completes_with_responses(self):
"""Answering the ask renders the prompt on the next round."""
mcp = self._context_prompt_server()
async with Client(mcp, mode="auto") as client:
ask = await client.session.get_prompt(
"summarize", allow_input_required=True
)
assert isinstance(ask, InputRequiredResult)
done = await client.session.get_prompt(
"summarize",
input_responses={
"context": mcp_types.ElicitResult(
action="accept", content={"context": "quarterly report"}
)
},
)
assert done.messages[0].content.text == ("Summarizing with quarterly report")
async def test_prompt_guard_rejected_on_handshake_era(self):
"""The result type only exists at 2026-07-28, so an older connection
gets the era named rather than a generic invalid-result failure."""
async with Client(self._context_prompt_server(), mode="legacy") as client:
with pytest.raises(MCPError, match="2026-07-28"):
await client.session.get_prompt("summarize")
class TestResourceGuard:
"""Resources and templates ask for input the same way tools and prompts do."""
@staticmethod
def _resource_server() -> FastMCP:
mcp = FastMCP("resource-guard")
@mcp.resource("data://report")
async def report(ctx: Context) -> str | InputRequiredResult:
responses = ctx.input_responses
if responses is None:
return _ask(
_elicit("context", "Which quarter?", "context"),
key="context",
request_state=None,
)
return f"Report for {_accepted(responses, 'context')['context']}"
@mcp.resource("data://report/{section}")
async def section_report(
section: str, ctx: Context
) -> str | InputRequiredResult:
responses = ctx.input_responses
if responses is None:
return _ask(
_elicit("context", f"Which quarter for {section}?", "context"),
key="context",
request_state=None,
)
quarter = _accepted(responses, "context")["context"]
return f"{section} for {quarter}"
return mcp
async def test_resource_emits_input_required(self):
async with Client(self._resource_server(), mode="auto") as client:
result = await client.session.read_resource(
"data://report", allow_input_required=True
)
assert isinstance(result, InputRequiredResult)
assert "context" in result.input_requests
async def test_resource_completes_with_responses(self):
async with Client(self._resource_server(), mode="auto") as client:
done = await client.session.read_resource(
"data://report",
input_responses={
"context": mcp_types.ElicitResult(
action="accept", content={"context": "Q3"}
)
},
)
assert done.contents[0].text == "Report for Q3"
async def test_resource_template_emits_input_required(self):
"""Templates share the converter, so the ask survives there too."""
async with Client(self._resource_server(), mode="auto") as client:
result = await client.session.read_resource(
"data://report/revenue", allow_input_required=True
)
assert isinstance(result, InputRequiredResult)
assert "context" in result.input_requests
async def test_resource_guard_rejected_on_handshake_era(self):
async with Client(self._resource_server(), mode="legacy") as client:
with pytest.raises(MCPError, match="2026-07-28"):
await client.session.read_resource("data://report")

232
tests/server/test_mrtr_guards_components.py Normal file
View file

@ -0,0 +1,232 @@
"""Guard-mode multi-round-trip for prompts and resources (SEP-2322).
`InputRequiredResult` is a *result type*, not a `tools/call` feature: any
request can resolve to one. A prompt or resource asks for client input exactly
the way a tool does — return the ask, read `ctx.input_responses` on the round
that follows.
These tests cover the emission side for prompts, concrete resources, and
resource templates, the 2026-07-28 era gate, and the proxy path, where the ask
must be forwarded to the parent rather than answered inside the proxy (a proxy
has no back-channel to the real user). Tool guards live in
``tests/server/test_mrtr_guards.py``.
"""
from __future__ import annotations
import mcp_types
import pytest
from mcp.shared.exceptions import MCPError
from mcp_types import ElicitRequest, InputRequiredResult
from fastmcp import Client, Context, FastMCP
from fastmcp.client.elicitation import ElicitResult
def _elicit(key: str, message: str, field: str) -> ElicitRequest:
"""A single-field form elicitation request keyed by ``key``."""
params = mcp_types.ElicitRequestFormParams(
message=message,
requested_schema={
"type": "object",
"properties": {field: {"type": "string"}},
"required": [field],
},
)
return ElicitRequest(method="elicitation/create", params=params)
def _ask(
request: ElicitRequest, key: str, request_state: str | None
) -> InputRequiredResult:
return InputRequiredResult(
result_type="input_required",
input_requests={key: request},
request_state=request_state,
)
def _accepted(responses: mcp_types.InputResponses, key: str) -> dict[str, object]:
"""The accepted form content for one answered elicitation."""
answer = responses[key]
assert isinstance(answer, mcp_types.ElicitResult)
assert answer.content is not None
return dict(answer.content)
def _modern_proxy(backend: FastMCP) -> FastMCP:
"""A proxy whose backend client negotiates the modern era, so the backend
can emit an `InputRequiredResult` for the proxy to round-trip."""
from fastmcp.server.providers.proxy import FastMCPProxy, ProxyClient
return FastMCPProxy(client_factory=lambda: ProxyClient(backend, mode="auto"))
class TestPromptGuard:
"""`InputRequiredResult` is a result type, not a tools/call feature, so a
prompt can ask for input the same way a tool does (SEP-2322)."""
@staticmethod
def _context_prompt_server() -> FastMCP:
mcp = FastMCP("prompt-guard")
@mcp.prompt
async def summarize(ctx: Context) -> str | InputRequiredResult:
responses = ctx.input_responses
if responses is None:
return _ask(
_elicit("context", "What context?", "context"),
key="context",
request_state=None,
)
return f"Summarizing with {_accepted(responses, 'context')['context']}"
return mcp
async def test_prompt_emits_input_required(self):
"""The asking round reaches the wire as an InputRequiredResult."""
async with Client(self._context_prompt_server(), mode="auto") as client:
result = await client.session.get_prompt(
"summarize", allow_input_required=True
)
assert isinstance(result, InputRequiredResult)
assert "context" in result.input_requests
async def test_prompt_completes_with_responses(self):
"""Answering the ask renders the prompt on the next round."""
mcp = self._context_prompt_server()
async with Client(mcp, mode="auto") as client:
ask = await client.session.get_prompt(
"summarize", allow_input_required=True
)
assert isinstance(ask, InputRequiredResult)
done = await client.session.get_prompt(
"summarize",
input_responses={
"context": mcp_types.ElicitResult(
action="accept", content={"context": "quarterly report"}
)
},
)
assert done.messages[0].content.text == ("Summarizing with quarterly report")
async def test_prompt_guard_rejected_on_handshake_era(self):
"""The result type only exists at 2026-07-28, so an older connection
gets the era named rather than a generic invalid-result failure."""
async with Client(self._context_prompt_server(), mode="legacy") as client:
with pytest.raises(MCPError, match="2026-07-28"):
await client.session.get_prompt("summarize")
class TestResourceGuard:
"""Resources and templates ask for input the same way tools and prompts do."""
@staticmethod
def _resource_server() -> FastMCP:
mcp = FastMCP("resource-guard")
@mcp.resource("data://report")
async def report(ctx: Context) -> str | InputRequiredResult:
responses = ctx.input_responses
if responses is None:
return _ask(
_elicit("context", "Which quarter?", "context"),
key="context",
request_state=None,
)
return f"Report for {_accepted(responses, 'context')['context']}"
@mcp.resource("data://report/{section}")
async def section_report(
section: str, ctx: Context
) -> str | InputRequiredResult:
responses = ctx.input_responses
if responses is None:
return _ask(
_elicit("context", f"Which quarter for {section}?", "context"),
key="context",
request_state=None,
)
quarter = _accepted(responses, "context")["context"]
return f"{section} for {quarter}"
return mcp
async def test_resource_emits_input_required(self):
async with Client(self._resource_server(), mode="auto") as client:
result = await client.session.read_resource(
"data://report", allow_input_required=True
)
assert isinstance(result, InputRequiredResult)
assert "context" in result.input_requests
async def test_resource_completes_with_responses(self):
async with Client(self._resource_server(), mode="auto") as client:
done = await client.session.read_resource(
"data://report",
input_responses={
"context": mcp_types.ElicitResult(
action="accept", content={"context": "Q3"}
)
},
)
assert done.contents[0].text == "Report for Q3"
async def test_resource_template_emits_input_required(self):
"""Templates share the converter, so the ask survives there too."""
async with Client(self._resource_server(), mode="auto") as client:
result = await client.session.read_resource(
"data://report/revenue", allow_input_required=True
)
assert isinstance(result, InputRequiredResult)
assert "context" in result.input_requests
async def test_resource_guard_rejected_on_handshake_era(self):
async with Client(self._resource_server(), mode="legacy") as client:
with pytest.raises(MCPError, match="2026-07-28"):
await client.session.read_resource("data://report")
class TestProxyForwarding:
"""A proxy forwards a backend guard's ask instead of answering it."""
async def test_guard_prompt_round_trips_through_proxy(self):
"""A guard prompt behind a proxy surfaces its ask instead of the proxy
trying to answer it. The proxy has no back-channel to the real user, so
driving the ask internally fails with "Elicitation not supported"."""
backend = TestPromptGuard._context_prompt_server()
async def answer(message, response_type, params, ctx):
return ElicitResult(
action="accept", content=response_type(context="quarterly report")
)
async with Client(
_modern_proxy(backend), mode="auto", elicitation_handler=answer
) as client:
result = await client.get_prompt("summarize")
assert result.messages[0].content.text == "Summarizing with quarterly report"
async def test_guard_resource_round_trips_through_proxy(self):
"""Concrete resources and templates forward the ask the same way."""
backend = TestResourceGuard._resource_server()
async def answer(message, response_type, params, ctx):
return ElicitResult(action="accept", content=response_type(context="Q3"))
async with Client(
_modern_proxy(backend), mode="auto", elicitation_handler=answer
) as client:
direct = await client.read_resource("data://report")
templated = await client.read_resource("data://report/revenue")
assert direct[0].text == "Report for Q3"
assert templated[0].text == "revenue for Q3"