bvandeusenandClaude Opus 5.5
3984f639e8
Second step of milestone 461. An MCP client (Claude Code, Claude Desktop via mcp-remote, Open WebUI) connects to /mcp with a personal API token. - mcp>=2.2 (SDK v2 MCPServer). Mounted by wrapping app.asgi_app, so /mcp never enters Quart's pipeline; stateless streamable HTTP so a redeploy never strands a client on a session id the new process didn't issue. - Auth: Authorization: Bearer stw_… resolves to a Caller (user + effective role) in a ContextVar; 401 with WWW-Authenticate otherwise. The web pipeline's database gate runs first, so a DB restart answers 503 instead of failing inside a tool. - Registry, default-deny: register_tool(fn, role=, read_only=) is the only door; role and read_only are required. Tools above the caller's role, or from a disabled plugin, are neither listed nor callable. Undeclared arguments are refused (not silently dropped); a tool's ValueError reaches the agent as its reason; a DB failure mid-tool says so. - Plugins contribute tools via an optional register_mcp_tools(register) export, all-or-nothing, replaced on hot-reload. Documented in docs/plugins/writing-a-plugin.md. - First tool: whoami. - Tests drive app.asgi_app directly (Quart's test client never reaches /mcp); the integration test mints a real token against Postgres and runs initialize → tools/list → whoami → revoke → 401. Task #4966 "MCP-2 — /mcp transport". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
75 lines
2.9 KiB
Python
75 lines
2.9 KiB
Python
"""Integration: the /mcp endpoint end to end against real Postgres.
|
|
|
|
A token is minted through the real service, then MCP requests go through the
|
|
app's ASGI callable (the same path a client's HTTP request takes): the
|
|
database gate, bearer resolution and role filtering all run for real.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from tests.mcp._http import HEADERS as _HEADERS, asgi_request, rpc, rpc_result
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
_NEEDS_DB = pytest.mark.skipif(
|
|
not os.environ.get("STEWARD_DATABASE_URL"),
|
|
reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
if not os.environ.get("STEWARD_DATABASE_URL"):
|
|
pytest.skip("needs Postgres")
|
|
from steward.app import create_app
|
|
return create_app(testing=False)
|
|
|
|
|
|
@_NEEDS_DB
|
|
def test_mcp_over_http_with_a_minted_token(app):
|
|
from steward.core import api_tokens
|
|
from steward.models.users import User, UserRole
|
|
|
|
async def _go():
|
|
uid = str(uuid.uuid4())
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
s.add(User(id=uid, username=f"mcp-{uid[:8]}", email=f"{uid}@test",
|
|
password_hash="x", role=UserRole.admin, is_active=True))
|
|
async with app.db_sessionmaker() as s:
|
|
raw, token = await api_tokens.create_token(
|
|
s, user_id=uid, user_role=UserRole.admin, name="it", role=UserRole.viewer)
|
|
await s.commit()
|
|
|
|
auth = {**_HEADERS, "Authorization": f"Bearer {raw}"}
|
|
async with app.mcp_server.session_manager.run():
|
|
status, _, body = await asgi_request(app, "POST", "/mcp", auth, rpc("initialize", {
|
|
"protocolVersion": "2025-06-18", "capabilities": {},
|
|
"clientInfo": {"name": "it", "version": "0"}}))
|
|
assert status == 200, body
|
|
|
|
status, _, body = await asgi_request(app, "POST", "/mcp", auth, rpc("tools/list", id_=2))
|
|
assert status == 200, body
|
|
assert "whoami" in {t["name"] for t in rpc_result(body)["result"]["tools"]}
|
|
|
|
status, _, body = await asgi_request(app, "POST", "/mcp", auth, rpc(
|
|
"tools/call", {"name": "whoami", "arguments": {}}, id_=3))
|
|
assert status == 200, body
|
|
me = json.loads(rpc_result(body)["result"]["content"][0]["text"])
|
|
# The token's viewer ceiling wins over the user's admin role.
|
|
assert (me["user_id"], me["role"], me["token_id"]) == (uid, "viewer", token.id)
|
|
|
|
async with app.db_sessionmaker() as s:
|
|
await api_tokens.revoke_token(s, user_id=uid, token_id=token.id)
|
|
await s.commit()
|
|
status, _, _ = await asgi_request(app, "POST", "/mcp", auth, rpc("tools/list", id_=4))
|
|
assert status == 401
|
|
|
|
asyncio.run(_go())
|