tegwick
b4b1dc1c7b
Add qonto_org_summary, qonto_list_transactions, and qonto_cost_run_rate_hints MCP tools, all routed through CapabilityService with protocol="mcp" -- same PolicyEngine.decide() path as REST, same deny-reason vocabulary. Skip snapshot_bundle as an MCP tool (REST already covers the composite read; not a separate privilege). CapabilityService now threads protocol through _execute/_emit_audit instead of hardcoding "rest". cost_run_rate_hints gets its own service method since it's an independent policy capability, not only a snapshot sub-field. Actor identity reuses REST's X-Actor-* header convention via a shared auth.actor_claims_from_headers(), read from the MCP Context's request when present. Fixed streamable_http_path defaulting to "/mcp", which doubled to "/mcp/mcp" once mounted under the "/mcp" prefix. Verified end-to-end with the mcp SDK's streamablehttp_client against the live fixture-backed server: tool list, allow/deny paths, and X-Actor-ID flowing through to the audit log exactly like REST. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
119 lines
4.4 KiB
Python
119 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from mcp.server.fastmcp import Context, FastMCP
|
|
from starlette.applications import Starlette
|
|
|
|
from qonto_assistant import __version__
|
|
from qonto_assistant.auth import actor_claims_from_headers
|
|
from qonto_assistant.config import Settings
|
|
from qonto_assistant.service import CapabilityService
|
|
|
|
|
|
def create_mcp_server(*, settings: Settings, service: CapabilityService | None = None) -> FastMCP:
|
|
"""Build the MCP adapter on the same capability core as the REST surface.
|
|
|
|
Every tool here maps 1:1 to a REST-allowed capability id and goes through
|
|
`CapabilityService`, which in turn calls the shared
|
|
`PolicyEngine.decide()` -- no separate MCP policy path. Client auth is
|
|
still the simple X-Actor-* header convention (QONTO-WP-0003-T03 upgrades
|
|
this to real OIDC/workload auth); no bank secrets are ever exposed here.
|
|
"""
|
|
server = FastMCP(
|
|
name=settings.service_name,
|
|
instructions=(
|
|
"Governed read-only Qonto finance awareness for Binky. "
|
|
"No spend, transfer, card, or volume-cost tools are exposed here."
|
|
),
|
|
stateless_http=True,
|
|
# Mounted at "/mcp" by the parent FastAPI app (see app.py); keep this
|
|
# sub-app's own route at root so the final path is "/mcp", not
|
|
# "/mcp/mcp".
|
|
streamable_http_path="/",
|
|
)
|
|
|
|
@server.tool()
|
|
def qonto_ping() -> dict[str, str]:
|
|
"""Smoke tool confirming the MCP adapter is reachable. No bank call."""
|
|
return {"status": "ok", "service": settings.service_name, "version": __version__}
|
|
|
|
if service is not None:
|
|
|
|
@server.tool()
|
|
def qonto_org_summary(ctx: Context) -> dict[str, Any]:
|
|
"""Organization name and per-account balances (org_summary capability)."""
|
|
claims = _claims(ctx, settings)
|
|
return service.get_accounts(claims=claims, request_id=_request_id(ctx), protocol="mcp")
|
|
|
|
@server.tool()
|
|
def qonto_list_transactions(
|
|
ctx: Context,
|
|
account_slug: str | None = None,
|
|
page: int = 1,
|
|
page_size: int = 50,
|
|
window_days: int = 31,
|
|
status: str | None = "completed",
|
|
side: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Capped, filtered transaction history (list_transactions capability)."""
|
|
claims = _claims(ctx, settings)
|
|
return service.list_transactions(
|
|
claims=claims,
|
|
request_id=_request_id(ctx),
|
|
account_slug=account_slug,
|
|
page=page,
|
|
page_size=page_size,
|
|
window_days=window_days,
|
|
status=status,
|
|
side=side,
|
|
protocol="mcp",
|
|
)
|
|
|
|
@server.tool()
|
|
def qonto_cost_run_rate_hints(
|
|
ctx: Context,
|
|
window_days: int = 90,
|
|
page_size: int = 50,
|
|
) -> dict[str, Any]:
|
|
"""Normalized recurring-cost hints, not a raw export (cost_run_rate_hints capability)."""
|
|
claims = _claims(ctx, settings)
|
|
return service.get_cost_run_rate_hints(
|
|
claims=claims,
|
|
request_id=_request_id(ctx),
|
|
window_days=window_days,
|
|
page_size=page_size,
|
|
protocol="mcp",
|
|
)
|
|
|
|
return server
|
|
|
|
|
|
def mcp_asgi_app(*, settings: Settings, service: CapabilityService | None = None) -> Starlette:
|
|
return create_mcp_server(settings=settings, service=service).streamable_http_app()
|
|
|
|
|
|
def _claims(ctx: Context, settings: Settings):
|
|
headers = _headers_from_context(ctx)
|
|
return actor_claims_from_headers(headers, settings)
|
|
|
|
|
|
def _headers_from_context(ctx: Context) -> Mapping[str, str]:
|
|
# request_context is unset outside a live MCP request (e.g. stdio
|
|
# transport, or a tool invoked directly in tests) -- fall back to the
|
|
# same default-actor behavior REST uses for header-less callers.
|
|
try:
|
|
request_context = ctx.request_context
|
|
except ValueError:
|
|
return {}
|
|
request = getattr(request_context, "request", None)
|
|
headers = getattr(request, "headers", None)
|
|
return headers if headers is not None else {}
|
|
|
|
|
|
def _request_id(ctx: Context) -> str:
|
|
headers = _headers_from_context(ctx)
|
|
header_id = headers.get("x-request-id")
|
|
return header_id if header_id else str(uuid4())
|