feat(mcp): add HTTP request log middleware and hosted wiring (#373 stack 2/3)

docs/config.md

@@ -129,4 +129,4 @@ These variables load into `pipefy_mcp.McpSettings` (`settings.mcp`). TOML keys u
 | `PIPEFY_MCP_PORT` | `8000` | HTTP bind port. |
 | `PIPEFY_MCP_ALLOW_INSECURE_HTTP_BIND` | `false` | Escape hatch: lets the unauthenticated `local` profile serve HTTP on a non-loopback host, exposing the full tool surface with no inbound bearer. The `remote` profile never needs it. |
 | `PIPEFY_MCP_UNIFIED_ENVELOPE` | `true` | When true, migrated tools return `{success, data, ...}`. |
-| `PIPEFY_MCP_LOG_LEVEL` | `INFO` | Governs **both** hosted structured JSON lines on **stdout** and the FastMCP root logger (RichHandler text on **stderr**). Accepts `DEBUG`, `INFO`, `WARNING`, `ERROR`, or `CRITICAL` (case-insensitive). The server passes the resolved level to `FastMCP(log_level=...)` at construction, so this is the single operator knob for MCP logging. Some MCP tooling documents a separate `FASTMCP_LOG_LEVEL` env var; when using `pipefy-mcp-server`, prefer `PIPEFY_MCP_LOG_LEVEL` — it configures the root logger explicitly and keeps structured events on the same level. |
+| `PIPEFY_MCP_LOG_LEVEL` | `INFO` | Governs the FastMCP root logger (RichHandler text on **stderr**) only. Accepts `DEBUG`, `INFO`, `WARNING`, `ERROR`, or `CRITICAL` (case-insensitive). Hosted structured JSON request/tool lines use a dedicated logger pinned at `INFO` on stderr, so raising this knob to quiet text logs does **not** drop those debugging events. Prefer `PIPEFY_MCP_LOG_LEVEL` over any `FASTMCP_LOG_LEVEL` env var when using `pipefy-mcp-server`. Structured lines stay on stderr so they never share the stdio JSON-RPC stdout channel. |

packages/mcp/AGENTS.md

@@ -91,6 +91,33 @@ allowlist for a proxied deployment) is separate follow-up work; see
 inputs also still assume a loopback peer that shares the client's disk (remote-safe
 file inputs are separate follow-up work).
 
+## Hosted structured logging
+
+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.
+
+Wiring lives in `wire_hosted_observability` (`observability/wiring.py`): it calls
+`streamable_http_app()` once, attaches request middleware, and returns the Starlette app.
+`run_server` serves that app with uvicorn directly (`access_log=False`) so the
+structured request line replaces uvicorn's text access log.
+`configure_observability_logging` pins the dedicated structured logger at `INFO`
+independently of `PIPEFY_MCP_LOG_LEVEL` (which only governs FastMCP/root text
+logs), so quieting noisy text does not drop request/tool lines.
+
+The request logger is **pure-ASGI middleware** (`RequestLogMiddleware`), never
+Starlette `BaseHTTPMiddleware`: `BaseHTTPMiddleware` buffers the response body,
+which breaks long-lived Streamable HTTP / SSE streams. The pure-ASGI middleware
+only inspects `http.response.start` (status + headers) and passes the body through.
+`request_id` prefers inbound `x-request-id`, then `x-correlation-id`, and mints a
+UUID only when both are absent (or blank), so an upstream proxy can keep one id
+across service boundaries. Tool lines go through the same emitter builders as
+HTTP lines (`build_tool_call_event` / `emit_structured_event`).
+
 ## Tool registration
 
 Tools are registered **once, at construction** (via `_register_pipefy_tools` in
@@ -233,8 +260,8 @@ built-in structured logger (`observability/tool_log_middleware.py`) is seeded
 by default only under the `remote` profile. That is a default, not a capability
 boundary: per-call concerns like observability and downstream protection apply to
 any deployment (only per-user concerns are hosted-specific), so a local deployment
-can register its own middleware. The logger emits through the `logging` module
-(stderr), never stdout, which is the stdio transport's JSON-RPC stream.
+can register its own middleware. Tool lines use the same stderr JSON emitter as
+HTTP request lines (`emit_structured_event`), never stdout.
 
 The wrap targets a FastMCP internal and is tested against `mcp==1.25.0`; the
 install is idempotent per app (the sentinel is per handler, not a global). This is

packages/mcp/src/pipefy_mcp/observability/json_logging.py

@@ -1,4 +1,4 @@
-"""Pure builders and stdout JSON emitter for hosted structured log events."""
+"""Pure builders and stderr JSON emitter for hosted structured log events."""
 
 from __future__ import annotations
 
@@ -8,10 +8,15 @@
 from datetime import UTC, datetime
 from typing import Any, Literal
 
-ToolCallOutcome = Literal["ok", "error"]
+ToolCallOutcome = Literal["ok", "error", "cancelled", "elicitation"]
 
 OBSERVABILITY_LOGGER_NAME = "pipefy_mcp.observability.structured"
 
+# Structured events always emit at INFO on this logger. PIPEFY_MCP_LOG_LEVEL
+# governs the FastMCP root logger (text) only, so quieting noisy text logs does
+# not silently drop hosted request/tool lines.
+STRUCTURED_LOG_LEVEL = logging.INFO
+
 HTTP_REQUEST_EVENT_KEYS = frozenset(
     {
         "event",
@@ -37,6 +42,7 @@
         "duration_ms",
         "arg_keys",
         "request_id",
+        "client_id",
     }
 )
 
@@ -67,16 +73,20 @@ def normalize_log_level(log_level: str) -> int:
         ) from exc
 
 
-def configure_observability_logging(*, log_level: str) -> logging.Logger:
-    """Attach a stdout JSON-line handler that does not propagate to the root logger."""
+def configure_observability_logging() -> logging.Logger:
+    """Attach a stderr JSON-line handler pinned at INFO (``propagate=False``).
+
+    Does not take ``PIPEFY_MCP_LOG_LEVEL``: that knob configures FastMCP's root
+    logger only. Hosted structured lines stay at INFO so an operator can quiet
+    text logs without losing request/tool debugging events.
+    """
     logger = logging.getLogger(OBSERVABILITY_LOGGER_NAME)
     logger.handlers.clear()
 
-    handler = logging.StreamHandler(sys.stdout)
+    handler = logging.StreamHandler(sys.stderr)
     handler.setFormatter(logging.Formatter("%(message)s"))
-    level = normalize_log_level(log_level)
-    handler.setLevel(level)
-    logger.setLevel(level)
+    handler.setLevel(STRUCTURED_LOG_LEVEL)
+    logger.setLevel(STRUCTURED_LOG_LEVEL)
     logger.addHandler(handler)
     logger.propagate = False
     return logger
@@ -90,7 +100,7 @@ def reset_observability_logging() -> None:
 
 
 def emit_structured_event(event: dict[str, Any]) -> None:
-    """Emit one allowlisted event as a single JSON line on stdout."""
+    """Emit one allowlisted event as a single JSON line on stderr."""
     line = json.dumps(event, separators=(",", ":"))
     logging.getLogger(OBSERVABILITY_LOGGER_NAME).info(line)
 
@@ -138,6 +148,7 @@ def build_tool_call_event(
     duration_ms: int | float,
     arg_keys: list[str],
     request_id: str | None,
+    client_id: str | None = None,
     timestamp: datetime | None = None,
 ) -> dict[str, Any]:
     """Build the allowlisted dict for one tool-call log line."""
@@ -149,4 +160,5 @@ def build_tool_call_event(
         "duration_ms": duration_ms,
         "arg_keys": arg_keys,
         "request_id": request_id,
+        "client_id": client_id,
     }

packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py

@@ -0,0 +1,135 @@
+"""Pure-ASGI middleware that emits one structured JSON line per HTTP request."""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Awaitable, Callable, MutableMapping
+from typing import Any
+from uuid import uuid4
+
+from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
+
+from pipefy_mcp.auth.resource_server import PipefyAccessToken
+from pipefy_mcp.observability.json_logging import (
+    build_http_request_event,
+    emit_structured_event,
+)
+
+ASGIApp = Callable[
+    [
+        MutableMapping[str, Any],
+        Callable[..., Awaitable[Any]],
+        Callable[..., Awaitable[None]],
+    ],
+    Awaitable[None],
+]
+ASGIReceive = Callable[[], Awaitable[MutableMapping[str, Any]]]
+ASGISend = Callable[[MutableMapping[str, Any]], Awaitable[None]]
+
+MCP_SESSION_ID_HEADER = "mcp-session-id"
+REQUEST_ID_HEADERS = ("x-request-id", "x-correlation-id")
+
+
+def _header_from_scope(scope: MutableMapping[str, Any], name: str) -> str | None:
+    name_bytes = name.lower().encode("latin-1")
+    for key, value in scope.get("headers", []):
+        if key.lower() == name_bytes:
+            return value.decode("latin-1")
+    return None
+
+
+def _header_from_response(headers: list[tuple[bytes, bytes]], name: str) -> str | None:
+    name_bytes = name.lower().encode("latin-1")
+    for key, value in headers:
+        if key.lower() == name_bytes:
+            return value.decode("latin-1")
+    return None
+
+
+def resolve_request_id(scope: MutableMapping[str, Any]) -> str:
+    """Prefer an inbound correlation header; otherwise mint a server-side id.
+
+    Honors ``x-request-id`` first, then ``x-correlation-id``. Empty or
+    whitespace-only values are ignored so a proxy that sends a blank header
+    does not suppress generation.
+    """
+    for name in REQUEST_ID_HEADERS:
+        raw = _header_from_scope(scope, name)
+        if raw is None:
+            continue
+        candidate = raw.strip()
+        if candidate:
+            return candidate
+    return str(uuid4())
+
+
+def _client_ip(scope: MutableMapping[str, Any]) -> str | None:
+    client = scope.get("client")
+    if not client:
+        return None
+    return client[0]
+
+
+def _caller_identity(
+    scope: MutableMapping[str, Any],
+) -> tuple[str | None, str | None]:
+    user = scope.get("user")
+    if not isinstance(user, AuthenticatedUser):
+        return None, None
+    access_token = user.access_token
+    sub = access_token.sub if isinstance(access_token, PipefyAccessToken) else None
+    return sub, access_token.client_id
+
+
+class RequestLogMiddleware:
+    """Log one HTTP request as JSON without buffering the response body."""
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(
+        self,
+        scope: MutableMapping[str, Any],
+        receive: ASGIReceive,
+        send: ASGISend,
+    ) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        started_at = time.perf_counter()
+        request_id = resolve_request_id(scope)
+        scope.setdefault("state", {})["request_id"] = request_id
+
+        status: int | None = None
+        response_session_id: str | None = None
+        request_session_id = _header_from_scope(scope, MCP_SESSION_ID_HEADER)
+
+        async def send_wrapper(message: MutableMapping[str, Any]) -> None:
+            nonlocal status, response_session_id
+            if message["type"] == "http.response.start":
+                status = message["status"]
+                response_session_id = _header_from_response(
+                    message.get("headers", []),
+                    MCP_SESSION_ID_HEADER,
+                )
+            await send(message)
+
+        try:
+            await self.app(scope, receive, send_wrapper)
+        finally:
+            sub, client_id = _caller_identity(scope)
+            duration_ms = round((time.perf_counter() - started_at) * 1000, 3)
+            emit_structured_event(
+                build_http_request_event(
+                    method=scope["method"],
+                    path=scope["path"],
+                    status=status,
+                    duration_ms=duration_ms,
+                    client_ip=_client_ip(scope),
+                    session_id=request_session_id or response_session_id,
+                    request_id=request_id,
+                    sub=sub,
+                    client_id=client_id,
+                )
+            )

packages/mcp/src/pipefy_mcp/observability/tool_log_middleware.py

@@ -1,47 +1,33 @@
 """Structured per-tool-call logging as a tool-call middleware.
 
 The first consumer of the tool-call middleware chain (see
-:mod:`pipefy_mcp.core.tool_middleware`): it emits one JSON line per call with the
-tool name, outcome, duration, the caller's client id, argument key names, and a
-request id for correlation. Being the first consumer, it also exercises the
+:mod:`pipefy_mcp.core.tool_middleware`): it emits one JSON line per call via the
+hosted structured emitter (``build_tool_call_event`` /
+``emit_structured_event``). Being the first consumer, it also exercises the
 chain end to end.
 
 Privacy: never logs the bearer, and never logs argument values, only their
 (bounded) key names.
 
-Output goes through the standard ``logging`` module, which writes to stderr. It
-must never write to stdout: the stdio transport frames JSON-RPC over stdout, so a
-stray log line there corrupts the protocol.
-
-Line integrity depends on the process's root log handler, which this module does
-not configure. FastMCP's default ``RichHandler`` wraps at width 80 in a non-TTY,
-which would split an event across physical lines; and a deployment that
-reconfigures root logging before construction can drop these INFO lines. The
-hosted logging wiring (the structured-log emitter) owns installing a plain,
-non-wrapping formatter at INFO so the one-line contract holds; until then a host
-that cares must install one.
+Output goes through the dedicated observability logger on stderr. It must never
+write to stdout: the stdio transport frames JSON-RPC over stdout, so a stray log
+line there corrupts the protocol. Under HTTP, ``configure_observability_logging``
+pins that logger at INFO independently of ``PIPEFY_MCP_LOG_LEVEL``.
 """
 
 from __future__ import annotations
 
 import asyncio
-import json
-import logging
 import time
-from typing import Literal
 
 from mcp import UrlElicitationRequiredError, types
 
 from pipefy_mcp.core.tool_middleware import CallNext, ToolCallContext
-
-logger = logging.getLogger("pipefy_mcp.observability.tool_call")
-
-# "cancelled" (client disconnect) and "elicitation" (the tool is asking the client
-# to visit a URL) are normal control flow, kept out of "error" so an error-rate
-# alert is not tripped by them. A governance short-circuit still reads as "error":
-# it returns isError=True and is indistinguishable from a tool-reported failure at
-# the result boundary.
-ToolCallOutcome = Literal["ok", "error", "cancelled", "elicitation"]
+from pipefy_mcp.observability.json_logging import (
+    ToolCallOutcome,
+    build_tool_call_event,
+    emit_structured_event,
+)
 
 
 def _outcome(result: types.ServerResult) -> ToolCallOutcome:
@@ -54,11 +40,6 @@ def _outcome(result: types.ServerResult) -> ToolCallOutcome:
     return "error" if getattr(result.root, "isError", False) else "ok"
 
 
-def _emit(event: dict[str, object]) -> None:
-    """Write one structured line via logging (stderr), never stdout."""
-    logger.info(json.dumps(event, separators=(",", ":")))
-
-
 async def tool_log_middleware(
     ctx: ToolCallContext, call_next: CallNext
 ) -> types.ServerResult:
@@ -90,18 +71,14 @@ async def tool_log_middleware(
         outcome = "error"
         raise
     finally:
-        # Guard here, not inside _emit, so the event dict is not built when the
-        # line would be discarded.
-        if logger.isEnabledFor(logging.INFO):
-            duration_ms = round((time.perf_counter() - started_at) * 1000, 3)
-            _emit(
-                {
-                    "event": "tool_call",
-                    "tool": ctx.tool_name,
-                    "outcome": outcome,
-                    "duration_ms": duration_ms,
-                    "arg_keys": list(ctx.argument_keys),
-                    "client_id": ctx.identity.client_id,
-                    "request_id": ctx.request_id,
-                }
+        duration_ms = round((time.perf_counter() - started_at) * 1000, 3)
+        emit_structured_event(
+            build_tool_call_event(
+                tool=ctx.tool_name,
+                outcome=outcome,
+                duration_ms=duration_ms,
+                arg_keys=list(ctx.argument_keys),
+                request_id=ctx.request_id,
+                client_id=ctx.identity.client_id,
             )
+        )

packages/mcp/src/pipefy_mcp/observability/wiring.py

@@ -0,0 +1,21 @@
+"""Wire hosted observability into the Streamable HTTP Starlette app."""
+
+from __future__ import annotations
+
+from mcp.server.fastmcp import FastMCP
+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:
+    """Build the HTTP app once and attach hosted observability.
+
+    ``streamable_http_app()`` must run exactly once: each call builds a new
+    Starlette app (only the session manager is cached), so calling it again after
+    wiring would drop the middleware.
+    """
+    http_app = app.streamable_http_app()
+    http_app.add_middleware(RequestLogMiddleware)
+    return http_app