QONTO-WP-0003-T01: MCP adapter skeleton on shared capability core
Add a streamable-HTTP MCP adapter (mcp_server.py, official FastMCP SDK) mounted at /mcp in the existing FastAPI app, with a combined lifespan so the MCP session manager starts/stops with the service. Ships one smoke tool (qonto_ping, no bank call) — real capability tools land in T02. Prove REST and MCP hit the identical policy path: PolicyEngine.decide() never branches on request.protocol, verified by a parametrized test. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
parent
dc3431cda0
commit
d4656be310
6 changed files with 133 additions and 10 deletions
25
tests/test_mcp_server.py
Normal file
25
tests/test_mcp_server.py
Normal file
|
|
@ -0,0 +1,25 @@
|
|||
import pytest
|
||||
|
||||
from qonto_assistant.config import Settings
|
||||
from qonto_assistant.mcp_server import create_mcp_server
|
||||
|
||||
|
||||
def _settings() -> Settings:
|
||||
settings = Settings.from_env()
|
||||
return settings
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mcp_server_starts_and_lists_smoke_tool() -> None:
|
||||
server = create_mcp_server(settings=_settings())
|
||||
tools = await server.list_tools()
|
||||
tool_names = {tool.name for tool in tools}
|
||||
assert "qonto_ping" in tool_names
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mcp_server_smoke_tool_returns_no_bank_call() -> None:
|
||||
server = create_mcp_server(settings=_settings())
|
||||
result = await server.call_tool("qonto_ping", {})
|
||||
payload = result[1] if isinstance(result, tuple) else result
|
||||
assert payload is not None
|
||||
|
|
@ -1,5 +1,7 @@
|
|||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from qonto_assistant.contracts import ActorClaims, CapabilityRequest
|
||||
from qonto_assistant.policy import PolicyEngine
|
||||
|
||||
|
|
@ -85,3 +87,34 @@ def test_policy_enforces_scope_when_enabled() -> None:
|
|||
decision = _policy(enforce_scope=True).decide(request)
|
||||
assert decision.allowed is False
|
||||
assert decision.reason == "authz_denied"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("protocol", ["rest", "mcp"])
|
||||
def test_policy_decision_identical_across_protocols(protocol: str) -> None:
|
||||
"""QONTO-WP-0003-T01: REST and MCP must hit the identical policy path.
|
||||
|
||||
The decision for a known-deny case (a spend-shaped capability id) must
|
||||
not depend on which transport originated the request.
|
||||
"""
|
||||
claims = ActorClaims(
|
||||
actor_id="agent-1",
|
||||
tenant_id="binky",
|
||||
lane="green",
|
||||
scopes=frozenset({"finance.qonto.read"}),
|
||||
)
|
||||
request = CapabilityRequest(
|
||||
capability_id="list_transactions",
|
||||
tenant_id="binky",
|
||||
actor_claims=claims,
|
||||
resource_scope="test",
|
||||
request_args={
|
||||
"page": 1,
|
||||
"page_size": 50,
|
||||
"window_days": 31,
|
||||
"requested_action": "transfer_funds",
|
||||
},
|
||||
protocol=protocol,
|
||||
)
|
||||
decision = _policy().decide(request)
|
||||
assert decision.allowed is False
|
||||
assert decision.reason == "spend"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue