diff --git a/CHANGELOG.md b/CHANGELOG.md index 1dbf8e52..8d971193 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ### Added +- **MCP**: hosted structured logging on the HTTP transport (#373). Each HTTP request and each tool call emits one JSON line on **stderr** (`event`: `http_request` / `tool_call`) for hosted debugging, with a privacy allowlist (method, path, status, duration, caller identity, per-tool `arg_keys` (key names only, never argument values)). Request logging uses pure-ASGI middleware; tool logging uses the `#378` tool-call middleware chain via the same structured emitter. The dedicated structured logger is pinned at `INFO` independently of `PIPEFY_MCP_LOG_LEVEL` (which governs FastMCP/root text only). Stdio is unchanged: stdout remains the JSON-RPC wire. HTTP is served via uvicorn with `access_log=False` so structured request lines replace uvicorn's text access log. - **CLI**: `pipefy relation table list --ids ` mirrors MCP `get_table_relations` (table-relation IDs, not database table IDs). - **CLI**: `pipefy pipe start-form ` mirrors MCP `get_start_form_fields` (optional `--required-only`). - **CLI**: `pipefy card fill --phase --fields '{"…"}'` fills phase fields non-interactively; filters to editable phase field IDs before `update_card`. Optional `--required-only` limits the phase field lookup to required fields. JSON responses may include `skipped_field_ids` when `--fields` keys are dropped by the filter. diff --git a/packages/mcp/AGENTS.md b/packages/mcp/AGENTS.md index 29327f2b..61019a7f 100644 --- a/packages/mcp/AGENTS.md +++ b/packages/mcp/AGENTS.md @@ -96,10 +96,13 @@ file inputs are separate follow-up work). The HTTP transport emits allowlisted JSON lines on stderr for hosted **debugging** (`pipefy_mcp/observability/`): one `http_request` line per request and one `tool_call` line per tool invocation (via `tool_log_middleware`). Fields are -privacy-bounded (no bearer, no argument values, no query string, no exception -messages). Stdio does **not** install the structured emitter: under stdio, -stdout is the JSON-RPC wire, and local installs should not arm that -process-global handler. +privacy-bounded: **excluded** are bearer tokens, argument values, query strings, +and exception messages; **included** on `http_request` lines are the caller's +`sub` and `client_id` when an authenticated bearer is present (attribution for +hosted debugging). `tool_call` lines deliberately omit `sub` until a consumer +needs it (see Tool-call middleware); they still carry `client_id` when available. +Stdio does **not** install the structured emitter: under stdio, stdout is the +JSON-RPC wire, and local installs should not arm that process-global handler. Wiring lives in `wire_hosted_observability` (`observability/wiring.py`): it calls `streamable_http_app()` once, attaches request middleware, and returns the Starlette app. diff --git a/packages/mcp/src/pipefy_mcp/observability/wiring.py b/packages/mcp/src/pipefy_mcp/observability/wiring.py index fd71b8d1..428a2463 100644 --- a/packages/mcp/src/pipefy_mcp/observability/wiring.py +++ b/packages/mcp/src/pipefy_mcp/observability/wiring.py @@ -6,10 +6,9 @@ from starlette.applications import Starlette from pipefy_mcp.observability.request_log_middleware import RequestLogMiddleware -from pipefy_mcp.settings import Settings -def wire_hosted_observability(app: FastMCP, _settings: Settings) -> Starlette: +def wire_hosted_observability(app: FastMCP) -> Starlette: """Build the HTTP app once and attach hosted observability. ``streamable_http_app()`` must run exactly once: each call builds a new diff --git a/packages/mcp/src/pipefy_mcp/server.py b/packages/mcp/src/pipefy_mcp/server.py index 8d321882..a9184cd2 100644 --- a/packages/mcp/src/pipefy_mcp/server.py +++ b/packages/mcp/src/pipefy_mcp/server.py @@ -150,7 +150,7 @@ async def _serve_streamable_http(app: FastMCP, settings: Settings) -> None: import uvicorn configure_observability_logging() - http_app = wire_hosted_observability(app, settings) + http_app = wire_hosted_observability(app) mcp = settings.mcp config = uvicorn.Config( http_app, diff --git a/packages/mcp/tests/observability/test_observability_end_to_end.py b/packages/mcp/tests/observability/test_observability_end_to_end.py new file mode 100644 index 00000000..b314c5d1 --- /dev/null +++ b/packages/mcp/tests/observability/test_observability_end_to_end.py @@ -0,0 +1,164 @@ +"""In-process end-to-end test for the hosted observability wiring. + +Drives a real ``wire_hosted_observability`` app (real session manager, real +stateful Streamable HTTP dispatch, real ``request_ctx`` propagation) through +``httpx.ASGITransport``: initialize, initialized notification, then two +sequential ``tools/call`` POSTs in the same session. This is the CI lock for +the D4 correlation design: the session task is spawned once at initialize, so +a contextvar-based correlation would stamp the initialize request id on every +tool line, and this test would fail. + +Tool lines come from ``tool_log_middleware`` (#378) via the shared structured +emitter, not a second CallToolRequest wrap. + +``json_response=True`` keeps responses as plain JSON (no SSE framing to +parse); the dispatch path under test, transport metadata plus per-message +``request_ctx``, is identical to the SSE-response mode. +""" + +from __future__ import annotations + +import json +from typing import Any + +import anyio +import httpx +import pytest +from mcp.server.fastmcp import FastMCP + +from pipefy_mcp.core.tool_middleware import install_tool_call_middleware +from pipefy_mcp.observability.json_logging import ( + configure_observability_logging, + reset_observability_logging, +) +from pipefy_mcp.observability.tool_log_middleware import tool_log_middleware +from pipefy_mcp.observability.wiring import wire_hosted_observability + +_ACCEPT = "application/json, text/event-stream" + + +@pytest.fixture(autouse=True) +def _isolated_observability_logger(): + reset_observability_logging() + yield + reset_observability_logging() + + +def _read_log_lines(capsys: pytest.CaptureFixture[str]) -> list[dict[str, Any]]: + return [ + json.loads(line) + for line in capsys.readouterr().err.splitlines() + if line.strip() + ] + + +def _initialize_body() -> dict[str, Any]: + return { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "obs-e2e", "version": "0"}, + }, + } + + +def _call_tool_body(request_id: int, text: str) -> dict[str, Any]: + return { + "jsonrpc": "2.0", + "id": request_id, + "method": "tools/call", + "params": {"name": "echo", "arguments": {"text": text}}, + } + + +@pytest.mark.anyio +async def test_two_calls_in_one_session_correlate_to_their_own_posts(capsys): + configure_observability_logging() + + app = FastMCP("obs-e2e") + app.settings.json_response = True + + @app.tool() + def echo(text: str) -> str: + return text + + install_tool_call_middleware(app, [tool_log_middleware]) + http_app = wire_hosted_observability(app) + + with anyio.fail_after(15): + async with http_app.router.lifespan_context(http_app): + transport = httpx.ASGITransport(app=http_app) + # Loopback base_url: the SDK's DNS-rebinding protection (default + # TransportSecuritySettings) rejects non-local Host headers with 421. + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8000" + ) as client: + init_response = await client.post( + "/mcp", + json=_initialize_body(), + headers={"accept": _ACCEPT}, + ) + assert init_response.status_code == 200 + session_id = init_response.headers["mcp-session-id"] + protocol_version = init_response.json()["result"]["protocolVersion"] + session_headers = { + "accept": _ACCEPT, + "mcp-session-id": session_id, + "mcp-protocol-version": protocol_version, + } + + notified = await client.post( + "/mcp", + json={"jsonrpc": "2.0", "method": "notifications/initialized"}, + headers=session_headers, + ) + assert notified.status_code == 202 + + first = await client.post( + "/mcp", + json=_call_tool_body(2, "first-call"), + headers=session_headers, + ) + assert first.status_code == 200 + assert first.json()["result"]["isError"] is False + + second = await client.post( + "/mcp", + json=_call_tool_body(3, "second-call"), + headers=session_headers, + ) + assert second.status_code == 200 + + lines = _read_log_lines(capsys) + http_lines = [line for line in lines if line["event"] == "http_request"] + tool_lines = [line for line in lines if line["event"] == "tool_call"] + + assert len(http_lines) == 4 # initialize, notification, two tool calls + assert len(tool_lines) == 2 + + http_request_ids = [line["request_id"] for line in http_lines] + initialize_request_id = http_request_ids[0] + + for tool_line in tool_lines: + assert tool_line["tool"] == "echo" + assert tool_line["outcome"] == "ok" + assert tool_line["arg_keys"] == ["text"] + # Correlation: the tool line carries the id of the POST that carried + # it, which the middleware also logged as an http_request line. + assert tool_line["request_id"] in http_request_ids + assert tool_line["request_id"] != initialize_request_id + + # The two calls belong to different POSTs, so their ids must differ; a + # contextvar frozen at session start would have made them equal. + assert tool_lines[0]["request_id"] != tool_lines[1]["request_id"] + + # Argument values never reach the structured stream. + serialized = json.dumps(lines) + assert "first-call" not in serialized + assert "second-call" not in serialized + + # Every line in the same session carries the session id. + assert {line["session_id"] for line in http_lines} == {session_id} diff --git a/packages/mcp/tests/observability/test_wiring.py b/packages/mcp/tests/observability/test_wiring.py index eb4b5f81..0ebd2485 100644 --- a/packages/mcp/tests/observability/test_wiring.py +++ b/packages/mcp/tests/observability/test_wiring.py @@ -1,45 +1,144 @@ -"""Tests for hosted observability wiring.""" +"""Tests for hosted observability wiring through the real auth stack. + +Unit tests inject ``scope["user"]`` directly and only prove extraction. These +tests drive ``wire_hosted_observability`` so AuthenticationMiddleware populates +``scope["user"]`` before RequestLogMiddleware's ``finally`` reads it, locking +middleware order and identity fields together. +""" from __future__ import annotations -from unittest.mock import MagicMock, patch +import json +from typing import Any +import anyio +import httpx import pytest +from mcp.server.auth.provider import AccessToken +from mcp.server.auth.settings import AuthSettings as FastMcpAuthSettings from mcp.server.fastmcp import FastMCP -from pipefy_auth import AuthSettings -from pipefy_sdk import PipefySettings +from starlette.applications import Starlette -from pipefy_mcp.observability.request_log_middleware import RequestLogMiddleware +from pipefy_mcp.auth.resource_server import PipefyAccessToken +from pipefy_mcp.observability.json_logging import ( + configure_observability_logging, + reset_observability_logging, +) from pipefy_mcp.observability.wiring import wire_hosted_observability -from pipefy_mcp.settings import McpSettings, Settings -_SETTINGS = Settings( - pipefy=PipefySettings(base_url="https://api.pipefy.com"), - auth=AuthSettings(), - mcp=McpSettings(transport="http"), -) +_ACCEPT = "application/json, text/event-stream" +_GOOD_TOKEN = "good-token" +_CLIENT_ID = "client-abc" +_SUB = "user-123" + + +class _StubTokenVerifier: + """Accept one fixed bearer and map it to a PipefyAccessToken with sub.""" + + async def verify_token(self, token: str) -> AccessToken | None: + if token != _GOOD_TOKEN: + return None + return PipefyAccessToken( + token=token, + client_id=_CLIENT_ID, + scopes=["read"], + expires_at=None, + resource="https://mcp.example.com/mcp", + sub=_SUB, + ) + + +@pytest.fixture(autouse=True) +def _isolated_observability_logger(): + reset_observability_logging() + yield + reset_observability_logging() + + +def _read_log_lines(capsys: pytest.CaptureFixture[str]) -> list[dict[str, Any]]: + return [ + json.loads(line) + for line in capsys.readouterr().err.splitlines() + if line.strip() + ] + + +def _build_auth_http_app() -> Starlette: + app = FastMCP( + "wiring-identity", + token_verifier=_StubTokenVerifier(), + auth=FastMcpAuthSettings( + issuer_url="https://issuer.example.com", + resource_server_url="https://mcp.example.com/mcp", + ), + ) + app.settings.json_response = True + return wire_hosted_observability(app) -@pytest.mark.unit -def test_wire_hosted_observability_calls_streamable_http_app_once() -> None: - app = FastMCP("test") - starlette_app = MagicMock() +@pytest.mark.anyio +async def test_wired_app_logs_null_identity_without_bearer(capsys): + """UnauthenticatedUser from the real auth stack yields null sub/client_id.""" + configure_observability_logging() + http_app = _build_auth_http_app() - with patch.object( - app, "streamable_http_app", return_value=starlette_app - ) as mock_app: - result = wire_hosted_observability(app, _SETTINGS) + with anyio.fail_after(10): + async with http_app.router.lifespan_context(http_app): + transport = httpx.ASGITransport(app=http_app) + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8000" + ) as client: + response = await client.post( + "/mcp", + json={"jsonrpc": "2.0", "id": 1, "method": "ping"}, + headers={"accept": _ACCEPT}, + ) - mock_app.assert_called_once_with() - assert result is starlette_app + assert response.status_code == 401 + lines = [ + line for line in _read_log_lines(capsys) if line.get("event") == "http_request" + ] + assert len(lines) == 1 + assert lines[0]["sub"] is None + assert lines[0]["client_id"] is None + assert lines[0]["status"] == 401 -@pytest.mark.unit -def test_wire_hosted_observability_adds_request_log_middleware() -> None: - app = FastMCP("test") - starlette_app = MagicMock() +@pytest.mark.anyio +async def test_wired_app_logs_sub_and_client_id_from_auth_stack(capsys): + """AuthenticatedUser set by BearerAuthBackend reaches RequestLogMiddleware.""" + configure_observability_logging() + http_app = _build_auth_http_app() - with patch.object(app, "streamable_http_app", return_value=starlette_app): - wire_hosted_observability(app, _SETTINGS) + with anyio.fail_after(10): + async with http_app.router.lifespan_context(http_app): + transport = httpx.ASGITransport(app=http_app) + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8000" + ) as client: + response = await client.post( + "/mcp", + json={ + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "wiring-identity", "version": "0"}, + }, + }, + headers={ + "accept": _ACCEPT, + "authorization": f"Bearer {_GOOD_TOKEN}", + }, + ) - starlette_app.add_middleware.assert_called_once_with(RequestLogMiddleware) + assert response.status_code == 200 + lines = [ + line for line in _read_log_lines(capsys) if line.get("event") == "http_request" + ] + assert len(lines) == 1 + assert lines[0]["sub"] == _SUB + assert lines[0]["client_id"] == _CLIENT_ID + assert _GOOD_TOKEN not in json.dumps(lines) diff --git a/packages/mcp/tests/test_server.py b/packages/mcp/tests/test_server.py index 6d4a3930..1d4bd7f9 100644 --- a/packages/mcp/tests/test_server.py +++ b/packages/mcp/tests/test_server.py @@ -389,7 +389,7 @@ async def test_serve_streamable_http_disables_uvicorn_access_log(remote_rs_env): await _serve_streamable_http(fake_app, settings) mock_configure.assert_called_once_with() - mock_wire.assert_called_once_with(fake_app, settings) + mock_wire.assert_called_once_with(fake_app) mock_config_cls.assert_called_once_with( mock_http_app, host="127.0.0.1",