feat(mcp): add tool-call structured logger (#373 stack 3/3)

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 <id,...>` mirrors MCP `get_table_relations` (table-relation IDs, not database table IDs).
 - **CLI**: `pipefy pipe start-form <pipe_id>` mirrors MCP `get_start_form_fields` (optional `--required-only`).
 - **CLI**: `pipefy card fill <card_id> --phase <phase_id> --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.

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.

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

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,

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}

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)

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",