Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MCP: mcp 2.3 (MCPServer, transport arguments on streamable_http_app, bearer read in middleware); tests for the 2026-07-28 revision and the resource check

Browse Source 
Claude-Session: https://claude.ai/code/session_01SSi5AvZ7pFmySK8zy3NPMh
This commit is contained in:
stroblme
2026-10-28 09:21:01 +01:00
parent 32eb8a327a
commit d1ff0e051d
8 changed files with 224 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/ai/agent.py
+5 -6
View File
@@ -20,7 +20,7 @@ from collections.abc import AsyncIterator, Awaitable, Callable
from typing import Any
import httpx
from mcp.types import TextContent
from mcp.types import CallToolResult, TextContent
from fluksio.ai import client
from fluksio.ai.config import AIConfig
@@ -68,7 +68,7 @@ async def tool_specs() -> list[dict[str, Any]]:
"function": {
"name": tool.name,
"description": tool.description or "",
"parameters": tool.inputSchema,
"parameters": tool.input_schema,
},
}
for tool in await mcp.list_tools()
@@ -108,10 +108,9 @@ async def _call_tool(name: str, arguments: dict[str, Any]) -> str:
result = await mcp.call_tool(name, arguments)
except Exception as exc: # noqa: BLE001 - an unknown tool, or refused arguments
return json.dumps({"error": str(exc)})
if isinstance(result, dict):
text = json.dumps(result)
else:
text = "".join(c.text for c in result if isinstance(c, TextContent))
if not isinstance(result, CallToolResult):
return json.dumps({"error": "The tool asked for more input than a chat gives."})
text = "".join(c.text for c in result.content if isinstance(c, TextContent))
if len(text) > _RESULT_LIMIT:
text = text[:_RESULT_LIMIT] + "… (truncated)"
return text
backend/fluksio/mcp/http.py
+7 -4
View File
@@ -167,18 +167,21 @@ class _ThroughThePortal:
def build_http_app(app: object) -> ASGIApp:
"""The MCP sub-app, wired to this app's API and its OAuth server."""
server.mcp.settings.stateless_http = True
server.mcp.settings.json_response = True
server.mcp.settings.auth = AuthSettings(
issuer_url=AnyHttpUrl(settings.oauth_issuer),
resource_server_url=AnyHttpUrl(settings.mcp_resource),
required_scopes=None,
validate_token_resource=True,
)
server.mcp.settings.transport_security = _transport_security()
# The SDK only takes a verifier through its constructor, and the instance is
# module-level so the tools can be declared at import time.
server.mcp._token_verifier = _JWTVerifier() # noqa: SLF001
server.connect(app)
return _ThroughThePortal(server.mcp.streamable_http_app())
return _ThroughThePortal(
server.mcp.streamable_http_app(
stateless_http=True,
json_response=True,
transport_security=_transport_security(),
)
)
backend/fluksio/mcp/server.py
+33 -26
View File
@@ -13,24 +13,46 @@ identity rather than some service account.
from __future__ import annotations
import logging
import warnings
from contextvars import ContextVar
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
from mcp.server.lowlevel.server import request_ctx
from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
from mcp.server.mcpserver import MCPServer
from fluksio.sdk.client import seg as _seg
logger = logging.getLogger(__name__)
# FastMCP's own settings model leaves `lifespan` a forward reference, which
# pydantic-settings warns about on every start. Nothing reads that field from
# the environment, so the warning says nothing about this app.
with warnings.catch_warnings():
warnings.filterwarnings("ignore", message="Field 'lifespan' has an incomplete")
mcp = FastMCP("fluksio")
#: The ``Authorization`` value tool calls carry onto the REST API. The endpoint's
#: middleware sets it from each MCP request; the assistant sets it to the token of
#: the person chatting, for the length of their turn, since no MCP request is
#: involved there.
caller_authorization: ContextVar[str | None] = ContextVar(
"caller_authorization", default=None
)
async def _carry_authorization(
ctx: ServerRequestContext[Any, Any], call_next: CallNext
) -> HandlerResult:
"""Make the bearer token of an MCP request the caller's for its tool calls.
The transport attaches the originating HTTP request to every JSON-RPC
message, so the token is read per message rather than per connection.
"""
headers = getattr(ctx.request, "headers", None)
authorization = headers.get("authorization") if headers else None
if not authorization:
return await call_next(ctx)
token = caller_authorization.set(authorization)
try:
return await call_next(ctx)
finally:
caller_authorization.reset(token)
mcp = MCPServer("fluksio", middleware=[_carry_authorization])
#: The tool calls never leave the process, so the host is a label, not a route.
_INTERNAL_BASE = "http://fluksio-mcp.internal"
@@ -39,13 +61,6 @@ _INTERNAL_BASE = "http://fluksio-mcp.internal"
#: hop. Built once by :func:`connect`, for the endpoint and the assistant alike.
_client: httpx.AsyncClient | None = None
#: The ``Authorization`` value for tool calls made from inside this process,
#: where no MCP request carries one: the assistant sets it to the token of the
#: person chatting, for the length of their turn.
caller_authorization: ContextVar[str | None] = ContextVar(
"caller_authorization", default=None
)
def set_client(client: httpx.AsyncClient | None) -> None:
global _client
@@ -77,18 +92,10 @@ async def aclose() -> None:
async def forward_caller_auth(request: httpx.Request) -> None:
"""Carry the caller's bearer token onto the REST call.
Over the endpoint, the transport attaches the originating HTTP request to
every JSON-RPC message, and a tool handler runs inside that request's
context, so the token is read per message rather than per connection. A
call from inside the process has no such request and uses what
:data:`caller_authorization` holds.
A tool handler runs inside the context :data:`caller_authorization` was set
in, by the endpoint's middleware or by the assistant.
"""
authorization = caller_authorization.get()
try:
source = request_ctx.get().request
except LookupError:
source = None
authorization = getattr(source, "headers", {}).get("authorization") or authorization
if authorization:
request.headers["authorization"] = authorization
backend/pyproject.toml
+1 -1
View File
@@ -65,7 +65,7 @@ dependencies = [
# The `inject` node's cron. Imported where it is used, behind a message
# that cannot fire while this line is here.
"croniter>=1.3.0",
"mcp>=1.30,<2",
"mcp>=2.3,<3",
"fluksio-worker>=0.1,<0.2",
# The Modules screen installs node code's packages with it, into a venv of
# the user's own. Present in the image; a pip install would otherwise have
backend/tests/mcp/test_mcp_http.py
+68 -6
View File
@@ -51,9 +51,6 @@ def over_mcp(monkeypatch: pytest.MonkeyPatch):
from fluksio.mcp import http as mcp_http
from fluksio.mcp import server as mcp_server
# A FastMCP instance enters its session manager once, and another test in
# the run may have built one already.
monkeypatch.setattr(mcp_server.mcp, "_session_manager", None, raising=False)
mcp_app = mcp_http.build_http_app(app)
def run(block: Callable[[httpx.AsyncClient], Awaitable[Any]]) -> Any:
@@ -194,6 +191,34 @@ def test_an_agent_can_list_and_call_tools(
assert called.json()["result"]["isError"] is False
def test_a_token_for_another_resource_is_refused(
over_mcp,
client: TestClient,
superuser_token_headers: dict[str, str],
agent: uuid.UUID,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The SDK compares the resource the verifier reports with its own URL."""
from fluksio.mcp import http as mcp_http
verify = mcp_http._JWTVerifier.verify_token
async def for_elsewhere(self: Any, token: str) -> Any:
accepted = await verify(self, token)
return accepted.model_copy(update={"resource": "https://other.example/mcp"})
monkeypatch.setattr(mcp_http._JWTVerifier, "verify_token", for_elsewhere)
me = client.get(
f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers
).json()
token = mcp_token(uuid.UUID(me["id"]), agent)
async def block(http: httpx.AsyncClient) -> httpx.Response:
return await http.post("/mcp", **rpc(token, "tools/list"))
assert over_mcp(block).status_code == 401
def test_revoking_an_agent_stops_the_token_it_already_holds(
over_mcp,
client: TestClient,
@@ -278,7 +303,7 @@ def test_a_real_client_can_connect_and_call_a_tool(
) as http_client:
async with streamable_http_client(
f"{settings.oauth_issuer}/mcp", http_client=http_client
) as (read, write, _):
) as (read, write):
async with ClientSession(read, write) as session:
initialized = await session.initialize()
tools = await session.list_tools()
@@ -286,9 +311,13 @@ def test_a_real_client_can_connect_and_call_a_tool(
return initialized, tools, called
initialized, tools, called = over_mcp(block)
assert initialized.serverInfo.name == "fluksio"
assert initialized.server_info.name == "fluksio"
assert {"list_flows", "run_flow"} <= {tool.name for tool in tools.tools}
assert called.isError is False
assert called.is_error is False
# The tool reached the API as the agent's person: refused calls come back
# as an error object in the text, not as a failed call.
answer = json.loads(called.content[0].text)
assert not (isinstance(answer, dict) and "error" in answer), answer
@pytest.mark.parametrize("revision", ["2025-03-26", "2025-06-18", "2025-11-25"])
@@ -326,3 +355,36 @@ def test_clients_of_older_protocol_revisions_still_connect(
assert initialized.json()["result"]["protocolVersion"] == revision
assert listed.status_code == 200
assert "list_flows" in {tool["name"] for tool in listed.json()["result"]["tools"]}
def test_a_request_of_the_2026_07_28_revision_is_served(
over_mcp,
client: TestClient,
superuser_token_headers: dict[str, str],
agent: uuid.UUID,
) -> None:
"""Its requests carry their own protocol envelope and no handshake."""
me = client.get(
f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers
).json()
token = mcp_token(uuid.UUID(me["id"]), agent)
meta = {
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientCapabilities": {},
}
call = rpc(token, "tools/call", _meta=meta, name="list_flows", arguments={})
call["headers"] |= {
"MCP-Protocol-Version": "2026-07-28",
"Mcp-Method": "tools/call",
"Mcp-Name": "list_flows",
}
async def block(http: httpx.AsyncClient) -> httpx.Response:
return await http.post("/mcp", **call)
answer = over_mcp(block)
assert answer.status_code == 200
result = answer.json()["result"]
assert result["isError"] is False
# Carrying the agent's token on to the API, or the call is refused there.
assert "error" not in json.loads(result["content"][0]["text"])
backend/tests/mcp/test_mcp_tools.py
+33
View File
@@ -39,3 +39,36 @@ def test_a_name_cannot_climb_out_of_its_route() -> None:
"DELETE /api/v1/flows/%2E%2E",
"GET /api/v1/flows/%2E%2E/nodes/%2E%2E%2Fsource/source",
]
def test_the_assistant_calls_a_tool_as_the_person_chatting() -> None:
"""The in-process path: no MCP request, so the person's token is set by hand."""
from fluksio.ai import agent
seen: list[str | None] = []
def record(request: httpx.Request) -> httpx.Response:
seen.append(request.headers.get("authorization"))
return httpx.Response(200, json=[{"name": "home"}])
async def calls() -> tuple[list[dict], str]:
previous = server._client
mock = httpx.AsyncClient(
transport=httpx.MockTransport(record),
base_url="http://test",
event_hooks={"request": [server.forward_caller_auth]},
)
server.set_client(mock)
try:
specs = await agent.tool_specs()
answer = await agent.mcp_caller("Bearer person")("list_flows", {})
finally:
server.set_client(previous)
await mock.aclose()
return specs, answer
specs, answer = asyncio.run(calls())
assert {"list_flows", "run_flow"} <= {s["function"]["name"] for s in specs}
assert all(s["function"]["parameters"]["type"] == "object" for s in specs)
assert seen == ["Bearer person"]
assert "home" in answer