coulomb/qonto-assistant
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

QONTO-WP-0003-T02: MCP tool catalog on the shared capability core

Browse source 
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>
This commit is contained in:
tegwick 2026-07-22 21:48:09 +02:00
parent ba2612682f
commit b4b1dc1c7b
7 changed files with 284 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

21
src/qonto_assistant/auth.py
View file

@ -1,15 +1,26 @@
from __future__ import annotations
from collections.abc import Mapping
from fastapi import Request
from qonto_assistant.config import Settings
from qonto_assistant.contracts import ActorClaims
def actor_claims_from_request(request: Request, settings: Settings) -> ActorClaims:
actor_id = request.headers.get("x-actor-id", "anonymous")
tenant_id = request.headers.get("x-tenant-id", settings.default_tenant_id)
lane = request.headers.get("x-actor-lane", settings.default_actor_lane)
raw_scopes = request.headers.get("x-actor-scopes", "")
def actor_claims_from_headers(headers: Mapping[str, str], settings: Settings) -> ActorClaims:
"""Shared REST/MCP claims parsing so both transports enforce identical actor identity.
Real workload/OIDC auth for the MCP transport is QONTO-WP-0003-T03; until
then MCP callers use the same X-Actor-* header convention as REST.
"""
actor_id = headers.get("x-actor-id", "anonymous")
tenant_id = headers.get("x-tenant-id", settings.default_tenant_id)
lane = headers.get("x-actor-lane", settings.default_actor_lane)
raw_scopes = headers.get("x-actor-scopes", "")
scopes = frozenset(scope.strip() for scope in raw_scopes.split(",") if scope.strip())
return ActorClaims(actor_id=actor_id, tenant_id=tenant_id, lane=lane, scopes=scopes)
def actor_claims_from_request(request: Request, settings: Settings) -> ActorClaims:
return actor_claims_from_headers(request.headers, settings)