diff --git a/docs/config.md b/docs/config.md index e5b20528..67986a1e 100644 --- a/docs/config.md +++ b/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. | diff --git a/packages/mcp/AGENTS.md b/packages/mcp/AGENTS.md index d8a35555..29327f2b 100644 --- a/packages/mcp/AGENTS.md +++ b/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 diff --git a/packages/mcp/src/pipefy_mcp/observability/json_logging.py b/packages/mcp/src/pipefy_mcp/observability/json_logging.py index a33fbcf6..0e00c3d0 100644 --- a/packages/mcp/src/pipefy_mcp/observability/json_logging.py +++ b/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, } diff --git a/packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py b/packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py new file mode 100644 index 00000000..6bec0514 --- /dev/null +++ b/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, + ) + ) diff --git a/packages/mcp/src/pipefy_mcp/observability/tool_log_middleware.py b/packages/mcp/src/pipefy_mcp/observability/tool_log_middleware.py index 766167d0..59f06201 100644 --- a/packages/mcp/src/pipefy_mcp/observability/tool_log_middleware.py +++ b/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, ) + ) diff --git a/packages/mcp/src/pipefy_mcp/observability/wiring.py b/packages/mcp/src/pipefy_mcp/observability/wiring.py new file mode 100644 index 00000000..fd71b8d1 --- /dev/null +++ b/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 diff --git a/packages/mcp/src/pipefy_mcp/server.py b/packages/mcp/src/pipefy_mcp/server.py index 8f07fb50..8d321882 100644 --- a/packages/mcp/src/pipefy_mcp/server.py +++ b/packages/mcp/src/pipefy_mcp/server.py @@ -5,6 +5,7 @@ from collections.abc import AsyncIterator, Callable, Sequence from contextlib import AbstractAsyncContextManager, asynccontextmanager +import anyio from mcp.server.fastmcp import FastMCP from pipefy_mcp.core.runtime import McpRuntime @@ -14,6 +15,7 @@ ) from pipefy_mcp.observability.json_logging import configure_observability_logging from pipefy_mcp.observability.tool_log_middleware import tool_log_middleware +from pipefy_mcp.observability.wiring import wire_hosted_observability from pipefy_mcp.settings import Settings from pipefy_mcp.tools.registry import ToolRegistry from pipefy_mcp.tools.validation_envelope import install_pipefy_validation_envelope @@ -137,6 +139,30 @@ def build_pipefy_mcp_server( return app +async def _serve_streamable_http(app: FastMCP, settings: Settings) -> None: + """Serve Streamable HTTP with hosted observability middleware wired in. + + The structured emitter is configured here, not in :func:`run_server`, so the + stdio path never installs it. Structured lines go to stderr (not the JSON-RPC + stdout wire); keeping configuration off the stdio path still avoids arming a + process-global handler that local installs do not need. + """ + import uvicorn + + configure_observability_logging() + http_app = wire_hosted_observability(app, settings) + mcp = settings.mcp + config = uvicorn.Config( + http_app, + host=mcp.host, + port=mcp.port, + log_level=mcp.log_level.lower(), + access_log=False, + ) + server = uvicorn.Server(config) + await server.serve() + + def run_server(settings: Settings) -> None: """Run the Pipefy MCP server. The single serve entry point for both transports. @@ -164,7 +190,6 @@ def run_server(settings: Settings) -> None: profile has no resource server). They differ only in the transport ``run``. """ mcp = settings.mcp - configure_observability_logging(log_level=mcp.log_level) if mcp.transport == "stdio": logger.info("Starting Pipefy MCP server over stdio (profile=%s)", mcp.profile) @@ -186,4 +211,4 @@ def run_server(settings: Settings) -> None: # Bind safety is enforced at the settings boundary (McpSettings._enforce_bind_safety); # host/port arrive already vetted, so there is nothing to re-check here. - build_pipefy_mcp_server(settings).run("streamable-http") + anyio.run(_serve_streamable_http, build_pipefy_mcp_server(settings), settings) diff --git a/packages/mcp/src/pipefy_mcp/settings.py b/packages/mcp/src/pipefy_mcp/settings.py index 8d06af48..0f00d1fb 100644 --- a/packages/mcp/src/pipefy_mcp/settings.py +++ b/packages/mcp/src/pipefy_mcp/settings.py @@ -114,8 +114,9 @@ def settings_customise_sources( log_level: McpLogLevel = Field( default="INFO", description=( - "Log level for hosted structured JSON events on stdout and the " - "FastMCP root logger on stderr (env: PIPEFY_MCP_LOG_LEVEL). " + "Log level for the FastMCP root logger on stderr " + "(env: PIPEFY_MCP_LOG_LEVEL). Hosted structured JSON events use a " + "dedicated logger pinned at INFO and are not governed by this knob. " "Accepts DEBUG, INFO, WARNING, ERROR, or CRITICAL; normalized to " "uppercase." ), diff --git a/packages/mcp/tests/conftest.py b/packages/mcp/tests/conftest.py index b1ff4677..6a436214 100644 --- a/packages/mcp/tests/conftest.py +++ b/packages/mcp/tests/conftest.py @@ -39,7 +39,7 @@ def _reset_observability_logging_between_tests(): The observability logger is process-global. Any test that reaches ``configure_observability_logging`` (directly or through ``run_server``) - would otherwise leave a handler bound to that test's captured stdout, and a + would otherwise leave a handler bound to that test's captured stderr, and a later ``emit_structured_event`` in an unrelated test would write into a closed stream. """ diff --git a/packages/mcp/tests/observability/test_json_logging.py b/packages/mcp/tests/observability/test_json_logging.py index 91e76107..24fe4528 100644 --- a/packages/mcp/tests/observability/test_json_logging.py +++ b/packages/mcp/tests/observability/test_json_logging.py @@ -112,20 +112,23 @@ def test_has_expected_fields(self): "duration_ms": 15, "arg_keys": ["card_id"], "request_id": "req-456", + "client_id": None, } - @pytest.mark.parametrize("outcome", ["ok", "error"]) - def test_outcome_is_ok_or_error(self, outcome: ToolCallOutcome): + @pytest.mark.parametrize("outcome", ["ok", "error", "cancelled", "elicitation"]) + def test_outcome_is_closed_set(self, outcome: ToolCallOutcome): event = build_tool_call_event( tool="find_cards", outcome=outcome, duration_ms=3, arg_keys=["pipe_id", "title"], request_id="req-outcome", + client_id="client-azp", timestamp=_FIXED_TIMESTAMP, ) assert event["outcome"] == outcome + assert event["client_id"] == "client-azp" def test_request_id_null_when_uncorrelated(self): event = build_tool_call_event( @@ -138,6 +141,7 @@ def test_request_id_null_when_uncorrelated(self): ) assert event["request_id"] is None + assert event["client_id"] is None class TestObservabilityLoggingEmitter: @@ -147,8 +151,8 @@ def _isolated_logger(self): yield reset_observability_logging() - def test_emits_valid_json_one_liner_on_stdout(self, capsys): - configure_observability_logging(log_level="INFO") + def test_emits_valid_json_one_liner_on_stderr(self, capsys): + configure_observability_logging() event = build_http_request_event( method="POST", path="/mcp", @@ -164,24 +168,35 @@ def test_emits_valid_json_one_liner_on_stdout(self, capsys): emit_structured_event(event) - line = capsys.readouterr().out.strip() + captured = capsys.readouterr() + assert captured.out == "" + line = captured.err.strip() assert "\n" not in line assert json.loads(line) == event - def test_warning_level_suppresses_info_events(self, capsys): - configure_observability_logging(log_level="WARNING") - emit_structured_event( - build_tool_call_event( - tool="get_card", - outcome="ok", - duration_ms=1, - arg_keys=["card_id"], - request_id="req-muted", - timestamp=_FIXED_TIMESTAMP, + def test_structured_events_still_emit_when_root_level_is_warning(self, capsys): + """PIPEFY_MCP_LOG_LEVEL must not mute the dedicated structured logger.""" + root = logging.getLogger() + previous = root.level + root.setLevel(logging.WARNING) + try: + configure_observability_logging() + emit_structured_event( + build_tool_call_event( + tool="get_card", + outcome="ok", + duration_ms=1, + arg_keys=["card_id"], + request_id="req-pinned", + timestamp=_FIXED_TIMESTAMP, + ) ) - ) + finally: + root.setLevel(previous) - assert capsys.readouterr().out == "" + captured = capsys.readouterr() + assert captured.out == "" + assert json.loads(captured.err.strip())["request_id"] == "req-pinned" def test_does_not_propagate_to_root_logger(self): root = logging.getLogger() @@ -195,7 +210,7 @@ def emit(self, record: logging.LogRecord) -> None: root.addHandler(root_handler) root.setLevel(logging.DEBUG) try: - configure_observability_logging(log_level="INFO") + configure_observability_logging() emit_structured_event( build_http_request_event( method="GET", @@ -214,16 +229,18 @@ def emit(self, record: logging.LogRecord) -> None: finally: root.removeHandler(root_handler) - def test_handler_targets_stdout_explicitly(self): - configure_observability_logging(log_level="INFO") + def test_handler_targets_stderr_explicitly(self): + configure_observability_logging() logger = logging.getLogger(OBSERVABILITY_LOGGER_NAME) assert len(logger.handlers) == 1 - assert logger.handlers[0].stream is sys.stdout + assert logger.handlers[0].stream is sys.stderr + assert logger.handlers[0].level == logging.INFO + assert logger.level == logging.INFO assert logger.propagate is False def test_configure_twice_keeps_one_handler_and_one_line(self, capsys): - configure_observability_logging(log_level="INFO") - configure_observability_logging(log_level="INFO") + configure_observability_logging() + configure_observability_logging() emit_structured_event( build_tool_call_event( @@ -238,7 +255,7 @@ def test_configure_twice_keeps_one_handler_and_one_line(self, capsys): logger = logging.getLogger(OBSERVABILITY_LOGGER_NAME) assert len(logger.handlers) == 1 - assert len(capsys.readouterr().out.strip().splitlines()) == 1 + assert len(capsys.readouterr().err.strip().splitlines()) == 1 def test_normalize_log_level_rejects_unknown_name(self): with pytest.raises(ValueError, match="invalid log level"): diff --git a/packages/mcp/tests/observability/test_request_log_middleware.py b/packages/mcp/tests/observability/test_request_log_middleware.py new file mode 100644 index 00000000..18004840 --- /dev/null +++ b/packages/mcp/tests/observability/test_request_log_middleware.py @@ -0,0 +1,430 @@ +"""Tests for the pure-ASGI HTTP request log middleware.""" + +from __future__ import annotations + +import json +from collections.abc import Awaitable, Callable, MutableMapping +from typing import Any + +import anyio +import pytest +from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser +from mcp.server.auth.provider import AccessToken +from starlette.authentication import UnauthenticatedUser + +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.request_log_middleware import ( + RequestLogMiddleware, + resolve_request_id, +) + +ASGIApp = Callable[ + [ + MutableMapping[str, Any], + Callable[..., Awaitable[Any]], + Callable[..., Awaitable[None]], + ], + Awaitable[None], +] + + +def _http_scope(**overrides: Any) -> dict[str, Any]: + scope: dict[str, Any] = { + "type": "http", + "asgi": {"spec_version": "2.3", "version": "3.0"}, + "http_version": "1.1", + "method": "POST", + "path": "/mcp", + "raw_path": b"/mcp", + "query_string": b"token=secret", + "root_path": "", + "scheme": "http", + "headers": [], + "client": ("127.0.0.1", 54321), + "server": ("testserver", 80), + "state": {}, + } + scope.update(overrides) + return scope + + +async def _noop_receive() -> dict[str, Any]: + return {"type": "http.disconnect"} + + +async def _run_middleware( + app: ASGIApp, + scope: dict[str, Any], + *, + receive: Callable[[], Awaitable[dict[str, Any]]] | None = None, +) -> tuple[list[dict[str, Any]], dict[str, Any]]: + sent: list[dict[str, Any]] = [] + + async def recording_send(message: dict[str, Any]) -> None: + sent.append(message) + + middleware = RequestLogMiddleware(app) + await middleware(scope, receive or _noop_receive, recording_send) + return sent, scope + + +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() + ] + + +@pytest.fixture(autouse=True) +def _isolated_observability_logger(): + reset_observability_logging() + yield + reset_observability_logging() + + +def _configure_for_capture() -> None: + configure_observability_logging() + + +@pytest.mark.anyio +async def test_emits_one_json_line_per_http_request(capsys): + _configure_for_capture() + + async def app(scope, receive, send): + await send({"type": "http.response.start", "status": 201, "headers": []}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + scope = _http_scope() + await _run_middleware(app, scope) + + lines = _read_log_lines(capsys) + assert len(lines) == 1 + assert lines[0]["event"] == "http_request" + assert lines[0]["method"] == "POST" + assert lines[0]["path"] == "/mcp" + assert lines[0]["status"] == 201 + assert lines[0]["client_ip"] == "127.0.0.1" + assert lines[0]["request_id"] == scope["state"]["request_id"] + assert "token=secret" not in json.dumps(lines[0]) + + +@pytest.mark.anyio +async def test_non_http_scope_passthrough_without_logging(capsys): + _configure_for_capture() + + async def app(scope, receive, send): + await send({"type": "lifespan.startup", "state": {}}) + + scope = {"type": "lifespan", "asgi": {"spec_version": "2.3", "version": "3.0"}} + await _run_middleware(app, scope) + + assert _read_log_lines(capsys) == [] + + +@pytest.mark.anyio +async def test_does_not_buffer_streaming_response_body(capsys): + _configure_for_capture() + first_chunk_sent = anyio.Event() + unblock_app = anyio.Event() + downstream_messages: list[dict[str, Any]] = [] + + async def streaming_app(scope, receive, send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send( + { + "type": "http.response.body", + "body": b"partial", + "more_body": True, + } + ) + first_chunk_sent.set() + await unblock_app.wait() + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + scope = _http_scope() + + async def recording_send(message: dict[str, Any]) -> None: + downstream_messages.append(message) + + middleware = RequestLogMiddleware(streaming_app) + with anyio.fail_after(5): + async with anyio.create_task_group() as tg: + tg.start_soon(middleware, scope, _noop_receive, recording_send) + await first_chunk_sent.wait() + body_messages = [ + message + for message in downstream_messages + if message["type"] == "http.response.body" + ] + assert body_messages[0]["body"] == b"partial" + unblock_app.set() + + assert len(_read_log_lines(capsys)) == 1 + + +@pytest.mark.anyio +async def test_disconnect_mid_stream_still_emits_one_line(capsys): + _configure_for_capture() + + async def streaming_app(scope, receive, send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send( + { + "type": "http.response.body", + "body": b"chunk", + "more_body": True, + } + ) + message = await receive() + assert message["type"] == "http.disconnect" + + async def disconnect_receive() -> dict[str, Any]: + return {"type": "http.disconnect"} + + await _run_middleware(streaming_app, _http_scope(), receive=disconnect_receive) + + lines = _read_log_lines(capsys) + assert len(lines) == 1 + assert lines[0]["status"] == 200 + + +@pytest.mark.anyio +async def test_send_failure_mid_stream_still_emits_one_line(capsys): + """A send that raises after the first chunk (aborted socket) still logs once.""" + _configure_for_capture() + sent_count = 0 + + async def streaming_app(scope, receive, send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"one", "more_body": True}) + await send({"type": "http.response.body", "body": b"two", "more_body": False}) + + async def failing_send(message: dict[str, Any]) -> None: + nonlocal sent_count + sent_count += 1 + if sent_count == 3: + raise RuntimeError("peer closed connection") + + middleware = RequestLogMiddleware(streaming_app) + with pytest.raises(RuntimeError, match="peer closed connection"): + await middleware(_http_scope(), _noop_receive, failing_send) + + lines = _read_log_lines(capsys) + assert len(lines) == 1 + assert lines[0]["status"] == 200 + + +@pytest.mark.anyio +async def test_crash_before_response_start_emits_status_null(capsys): + _configure_for_capture() + + async def crashing_app(scope, receive, send): + raise RuntimeError("boom before response.start") + + with pytest.raises(RuntimeError, match="boom before response.start"): + await _run_middleware(crashing_app, _http_scope()) + + lines = _read_log_lines(capsys) + assert len(lines) == 1 + assert lines[0]["status"] is None + + +@pytest.mark.anyio +async def test_request_id_is_stored_on_scope_state(capsys): + _configure_for_capture() + + async def app(scope, receive, send): + assert "request_id" in scope["state"] + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + scope = _http_scope() + await _run_middleware(app, scope) + + lines = _read_log_lines(capsys) + assert lines[0]["request_id"] == scope["state"]["request_id"] + + +@pytest.mark.anyio +async def test_request_id_from_x_request_id_header(capsys): + _configure_for_capture() + + async def app(scope, receive, send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + scope = _http_scope(headers=[(b"x-request-id", b"upstream-req-1")]) + await _run_middleware(app, scope) + + lines = _read_log_lines(capsys) + assert lines[0]["request_id"] == "upstream-req-1" + assert scope["state"]["request_id"] == "upstream-req-1" + + +@pytest.mark.anyio +async def test_request_id_falls_back_to_x_correlation_id(capsys): + _configure_for_capture() + + async def app(scope, receive, send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + scope = _http_scope(headers=[(b"x-correlation-id", b"corr-abc")]) + await _run_middleware(app, scope) + + lines = _read_log_lines(capsys) + assert lines[0]["request_id"] == "corr-abc" + + +@pytest.mark.anyio +async def test_x_request_id_wins_over_x_correlation_id(capsys): + _configure_for_capture() + + async def app(scope, receive, send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + scope = _http_scope( + headers=[ + (b"x-correlation-id", b"corr-loses"), + (b"x-request-id", b"req-wins"), + ] + ) + await _run_middleware(app, scope) + + lines = _read_log_lines(capsys) + assert lines[0]["request_id"] == "req-wins" + + +def test_resolve_request_id_ignores_blank_headers_and_mints(): + scope = _http_scope( + headers=[ + (b"x-request-id", b" "), + (b"x-correlation-id", b""), + ] + ) + minted = resolve_request_id(scope) + assert minted + assert minted.strip() == minted + + +@pytest.mark.anyio +async def test_sub_and_client_id_null_without_auth_context(capsys): + _configure_for_capture() + + async def app(scope, receive, send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + await _run_middleware(app, _http_scope()) + + lines = _read_log_lines(capsys) + assert lines[0]["sub"] is None + assert lines[0]["client_id"] is None + + +@pytest.mark.anyio +async def test_sub_and_client_id_from_scope_user(capsys): + _configure_for_capture() + access_token = PipefyAccessToken( + token="the-token", + client_id="client-abc", + scopes=["read"], + sub="user-123", + ) + + async def app(scope, receive, send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + scope = _http_scope( + user=AuthenticatedUser(access_token), + headers=[(b"authorization", b"Bearer the-token")], + ) + await _run_middleware(app, scope) + + lines = _read_log_lines(capsys) + assert lines[0]["sub"] == "user-123" + assert lines[0]["client_id"] == "client-abc" + assert "the-token" not in json.dumps(lines) + + +@pytest.mark.anyio +async def test_unauthenticated_user_yields_null_identity(capsys): + """The real auth stack sets UnauthenticatedUser, not a missing key.""" + _configure_for_capture() + + async def app(scope, receive, send): + await send({"type": "http.response.start", "status": 401, "headers": []}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + scope = _http_scope(user=UnauthenticatedUser()) + await _run_middleware(app, scope) + + lines = _read_log_lines(capsys) + assert lines[0]["sub"] is None + assert lines[0]["client_id"] is None + assert lines[0]["status"] == 401 + + +@pytest.mark.anyio +async def test_session_id_from_request_header(capsys): + _configure_for_capture() + + async def app(scope, receive, send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + scope = _http_scope( + headers=[(b"mcp-session-id", b"request-session")], + ) + await _run_middleware(app, scope) + + lines = _read_log_lines(capsys) + assert lines[0]["session_id"] == "request-session" + + +@pytest.mark.anyio +async def test_session_id_falls_back_to_response_header(capsys): + _configure_for_capture() + + async def app(scope, receive, send): + await send( + { + "type": "http.response.start", + "status": 200, + "headers": [(b"mcp-session-id", b"response-session")], + } + ) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + await _run_middleware(app, _http_scope()) + + lines = _read_log_lines(capsys) + assert lines[0]["session_id"] == "response-session" + + +@pytest.mark.anyio +async def test_plain_access_token_yields_null_sub(capsys): + _configure_for_capture() + access_token = AccessToken(token="the-token", client_id="client-abc", scopes=[]) + + async def app(scope, receive, send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + + scope = _http_scope( + user=AuthenticatedUser(access_token), + headers=[(b"authorization", b"Bearer the-token")], + ) + await _run_middleware(app, scope) + + lines = _read_log_lines(capsys) + assert lines[0]["sub"] is None + assert lines[0]["client_id"] == "client-abc" + assert "the-token" not in json.dumps(lines) diff --git a/packages/mcp/tests/observability/test_tool_log_middleware.py b/packages/mcp/tests/observability/test_tool_log_middleware.py index a340d486..8e2d789d 100644 --- a/packages/mcp/tests/observability/test_tool_log_middleware.py +++ b/packages/mcp/tests/observability/test_tool_log_middleware.py @@ -1,22 +1,25 @@ """Unit tests for the structured tool-call logging middleware. Pins the log shape and the two protocol-safety invariants: the outcome derives -from the call result, and output goes to logging (stderr), never stdout, so it -cannot corrupt the stdio JSON-RPC stream. +from the call result, and output goes through the hosted structured emitter on +stderr, never stdout, so it cannot corrupt the stdio JSON-RPC stream. """ from __future__ import annotations import asyncio import json -import logging -import sys import pytest from mcp import UrlElicitationRequiredError, types from pipefy_mcp.auth.request_identity import CallerIdentity from pipefy_mcp.core.tool_middleware import ToolCallContext +from pipefy_mcp.observability.json_logging import ( + TOOL_CALL_EVENT_KEYS, + configure_observability_logging, + reset_observability_logging, +) from pipefy_mcp.observability.tool_log_middleware import tool_log_middleware @@ -39,48 +42,68 @@ def _ok_result() -> types.ServerResult: ) -def _log_line(caplog) -> dict: - records = [ - r for r in caplog.records if r.name == "pipefy_mcp.observability.tool_call" +def _read_log_lines(capsys: pytest.CaptureFixture[str]) -> list[dict]: + return [ + json.loads(line) + for line in capsys.readouterr().err.splitlines() + if line.strip() ] - assert len(records) == 1 - return json.loads(records[0].getMessage()) + + +@pytest.fixture(autouse=True) +def _isolated_observability_logger(): + reset_observability_logging() + yield + reset_observability_logging() + + +def _configure_for_capture() -> None: + # Must run inside the test body so StreamHandler binds the capsys-redirected + # stderr, not the process's original stream. + configure_observability_logging() @pytest.mark.unit -def test_logs_one_line_with_the_documented_fields(caplog): +def test_logs_one_line_with_the_documented_fields(capsys): + _configure_for_capture() + async def terminal(ctx): return _ok_result() - with caplog.at_level(logging.INFO, logger="pipefy_mcp.observability.tool_call"): - asyncio.run(tool_log_middleware(_context(card_id="1"), terminal)) + asyncio.run(tool_log_middleware(_context(card_id="1"), terminal)) - event = _log_line(caplog) + lines = _read_log_lines(capsys) + assert len(lines) == 1 + event = lines[0] + assert set(event.keys()) == TOOL_CALL_EVENT_KEYS + assert event["event"] == "tool_call" assert event["tool"] == "get_card" assert event["outcome"] == "ok" assert event["arg_keys"] == ["card_id"] assert event["client_id"] == "acting-client" assert event["request_id"] == "req-42" assert isinstance(event["duration_ms"], (int, float)) + assert "timestamp" in event @pytest.mark.unit -def test_never_logs_argument_values_or_a_bearer(caplog): +def test_never_logs_argument_values_or_a_bearer(capsys): + _configure_for_capture() + async def terminal(ctx): return _ok_result() - with caplog.at_level(logging.INFO, logger="pipefy_mcp.observability.tool_call"): - asyncio.run( - tool_log_middleware(_context(token="super-secret-bearer"), terminal) - ) + asyncio.run(tool_log_middleware(_context(token="super-secret-bearer"), terminal)) - raw = caplog.records[-1].getMessage() + raw = capsys.readouterr().err assert "super-secret-bearer" not in raw - assert "token" in json.loads(raw)["arg_keys"] # the key is logged, not the value + assert "token" in json.loads(raw.strip())["arg_keys"] @pytest.mark.unit -def test_error_result_logs_outcome_error(caplog): +def test_error_result_logs_outcome_error(capsys): + _configure_for_capture() + async def terminal(ctx): return types.ServerResult( types.CallToolResult( @@ -88,43 +111,43 @@ async def terminal(ctx): ) ) - with caplog.at_level(logging.INFO, logger="pipefy_mcp.observability.tool_call"): - asyncio.run(tool_log_middleware(_context(card_id="1"), terminal)) + asyncio.run(tool_log_middleware(_context(card_id="1"), terminal)) - assert _log_line(caplog)["outcome"] == "error" + assert _read_log_lines(capsys)[0]["outcome"] == "error" @pytest.mark.unit -def test_cancellation_logs_outcome_cancelled_and_re_raises(caplog): +def test_cancellation_logs_outcome_cancelled_and_re_raises(capsys): """A client disconnect is control flow, not an error, and must propagate.""" + _configure_for_capture() async def terminal(ctx): raise asyncio.CancelledError - with caplog.at_level(logging.INFO, logger="pipefy_mcp.observability.tool_call"): - with pytest.raises(asyncio.CancelledError): - asyncio.run(tool_log_middleware(_context(card_id="1"), terminal)) + with pytest.raises(asyncio.CancelledError): + asyncio.run(tool_log_middleware(_context(card_id="1"), terminal)) - assert _log_line(caplog)["outcome"] == "cancelled" + assert _read_log_lines(capsys)[0]["outcome"] == "cancelled" @pytest.mark.unit -def test_elicitation_logs_outcome_elicitation_and_re_raises(caplog): +def test_elicitation_logs_outcome_elicitation_and_re_raises(capsys): """A URL-elicitation signal is a continuation, not an error, and must propagate.""" + _configure_for_capture() async def terminal(ctx): raise UrlElicitationRequiredError([]) - with caplog.at_level(logging.INFO, logger="pipefy_mcp.observability.tool_call"): - with pytest.raises(UrlElicitationRequiredError): - asyncio.run(tool_log_middleware(_context(card_id="1"), terminal)) + with pytest.raises(UrlElicitationRequiredError): + asyncio.run(tool_log_middleware(_context(card_id="1"), terminal)) - assert _log_line(caplog)["outcome"] == "elicitation" + assert _read_log_lines(capsys)[0]["outcome"] == "elicitation" @pytest.mark.unit -def test_other_propagated_exception_logs_outcome_error_and_re_raises(caplog): +def test_other_propagated_exception_logs_outcome_error_and_re_raises(capsys): """Any other propagating exception is logged as an error and re-raised.""" + _configure_for_capture() class Boom(Exception): pass @@ -132,29 +155,21 @@ class Boom(Exception): async def terminal(ctx): raise Boom("propagate me") - with caplog.at_level(logging.INFO, logger="pipefy_mcp.observability.tool_call"): - with pytest.raises(Boom, match="propagate me"): - asyncio.run(tool_log_middleware(_context(card_id="1"), terminal)) + with pytest.raises(Boom, match="propagate me"): + asyncio.run(tool_log_middleware(_context(card_id="1"), terminal)) - assert _log_line(caplog)["outcome"] == "error" + assert _read_log_lines(capsys)[0]["outcome"] == "error" @pytest.mark.unit def test_writes_nothing_to_stdout(capsys): - """The logger must never touch stdout: it is the stdio transport's JSON-RPC stream.""" + """The emitter must never touch stdout: it is the stdio transport's JSON-RPC stream.""" + _configure_for_capture() async def terminal(ctx): return _ok_result() - # Route logging to stderr explicitly, then assert stdout stays empty. - handler = logging.StreamHandler(sys.stderr) - log = logging.getLogger("pipefy_mcp.observability.tool_call") - log.addHandler(handler) - log.setLevel(logging.INFO) - try: - asyncio.run(tool_log_middleware(_context(card_id="1"), terminal)) - finally: - log.removeHandler(handler) + asyncio.run(tool_log_middleware(_context(card_id="1"), terminal)) captured = capsys.readouterr() assert captured.out == "" diff --git a/packages/mcp/tests/observability/test_wiring.py b/packages/mcp/tests/observability/test_wiring.py new file mode 100644 index 00000000..eb4b5f81 --- /dev/null +++ b/packages/mcp/tests/observability/test_wiring.py @@ -0,0 +1,45 @@ +"""Tests for hosted observability wiring.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from mcp.server.fastmcp import FastMCP +from pipefy_auth import AuthSettings +from pipefy_sdk import PipefySettings + +from pipefy_mcp.observability.request_log_middleware import RequestLogMiddleware +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"), +) + + +@pytest.mark.unit +def test_wire_hosted_observability_calls_streamable_http_app_once() -> None: + app = FastMCP("test") + starlette_app = MagicMock() + + with patch.object( + app, "streamable_http_app", return_value=starlette_app + ) as mock_app: + result = wire_hosted_observability(app, _SETTINGS) + + mock_app.assert_called_once_with() + assert result is starlette_app + + +@pytest.mark.unit +def test_wire_hosted_observability_adds_request_log_middleware() -> None: + app = FastMCP("test") + starlette_app = MagicMock() + + with patch.object(app, "streamable_http_app", return_value=starlette_app): + wire_hosted_observability(app, _SETTINGS) + + starlette_app.add_middleware.assert_called_once_with(RequestLogMiddleware) diff --git a/packages/mcp/tests/test_server.py b/packages/mcp/tests/test_server.py index bd0673cd..6d4a3930 100644 --- a/packages/mcp/tests/test_server.py +++ b/packages/mcp/tests/test_server.py @@ -1,6 +1,6 @@ import json from datetime import timedelta -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest @@ -17,6 +17,7 @@ from pipefy_mcp.server import ( _make_lifespan, _register_pipefy_tools, + _serve_streamable_http, build_pipefy_mcp_server, default_tool_middlewares, run_server, @@ -107,7 +108,8 @@ def test_build_pipefy_mcp_server_passes_log_level_to_fastmcp(mocked_runtime): @pytest.mark.unit -def test_run_server_configures_structured_logging(monkeypatch): +def test_run_server_stdio_does_not_configure_structured_logging(monkeypatch): + """Stdio must not install the structured emitter (HTTP-only configuration).""" monkeypatch.delenv("PIPEFY_MCP_PROFILE", raising=False) monkeypatch.delenv("PIPEFY_MCP_TRANSPORT", raising=False) with ( @@ -117,7 +119,7 @@ def test_run_server_configures_structured_logging(monkeypatch): run_server( resolve_mcp_settings(profile=None, transport=None, host=None, port=None) ) - mock_configure.assert_called_once_with(log_level="INFO") + mock_configure.assert_not_called() @pytest.mark.unit @@ -302,21 +304,23 @@ def test_run_server_local_profile_over_http_serves_without_inbound_auth(): (inbound bearer validation) is built even when serving over HTTP. """ fake_app = MagicMock() - with patch( - "pipefy_mcp.server.build_pipefy_mcp_server", return_value=fake_app - ) as mock_build: - run_server( - resolve_mcp_settings( - profile="local", transport="http", host="127.0.0.1", port=9200 - ) + with ( + patch( + "pipefy_mcp.server.build_pipefy_mcp_server", return_value=fake_app + ) as mock_build, + patch("pipefy_mcp.server.anyio.run") as mock_anyio_run, + ): + settings = resolve_mcp_settings( + profile="local", transport="http", host="127.0.0.1", port=9200 ) + run_server(settings) mock_build.assert_called_once() (built_settings,), _ = mock_build.call_args assert built_settings.mcp.profile == "local" assert built_settings.mcp.host == "127.0.0.1" assert built_settings.mcp.port == 9200 - fake_app.run.assert_called_once_with("streamable-http") + mock_anyio_run.assert_called_once_with(_serve_streamable_http, fake_app, settings) @pytest.mark.unit @@ -326,16 +330,18 @@ def test_run_server_remote_profile_defaults_to_http_transport( """``--profile remote`` with no ``--transport`` serves HTTP (profile-derived default).""" monkeypatch.delenv("PIPEFY_MCP_TRANSPORT", raising=False) fake_app = MagicMock() - with patch( - "pipefy_mcp.server.build_pipefy_mcp_server", return_value=fake_app - ) as mock_build: - run_server( - resolve_mcp_settings( - profile="remote", transport=None, host="127.0.0.1", port=9300 - ) + with ( + patch( + "pipefy_mcp.server.build_pipefy_mcp_server", return_value=fake_app + ) as mock_build, + patch("pipefy_mcp.server.anyio.run") as mock_anyio_run, + ): + settings = resolve_mcp_settings( + profile="remote", transport=None, host="127.0.0.1", port=9300 ) + run_server(settings) - fake_app.run.assert_called_once_with("streamable-http") + mock_anyio_run.assert_called_once_with(_serve_streamable_http, fake_app, settings) (built_settings,), _ = mock_build.call_args assert built_settings.mcp.profile == "remote" @@ -344,20 +350,54 @@ def test_run_server_remote_profile_defaults_to_http_transport( def test_run_server_http_builds_the_app_and_serves_over_streamable_http(remote_rs_env): """The remote HTTP path builds through the shared builder and serves streamable-http.""" fake_app = MagicMock() - with patch( - "pipefy_mcp.server.build_pipefy_mcp_server", return_value=fake_app - ) as mock_build: - run_server( - resolve_mcp_settings( - profile="remote", transport="http", host="127.0.0.1", port=9123 - ) + with ( + patch( + "pipefy_mcp.server.build_pipefy_mcp_server", return_value=fake_app + ) as mock_build, + patch("pipefy_mcp.server.anyio.run") as mock_anyio_run, + ): + settings = resolve_mcp_settings( + profile="remote", transport="http", host="127.0.0.1", port=9123 ) + run_server(settings) (built_settings,), _ = mock_build.call_args assert built_settings.mcp.profile == "remote" assert built_settings.mcp.host == "127.0.0.1" assert built_settings.mcp.port == 9123 - fake_app.run.assert_called_once_with("streamable-http") + mock_anyio_run.assert_called_once_with(_serve_streamable_http, fake_app, settings) + + +@pytest.mark.anyio +async def test_serve_streamable_http_disables_uvicorn_access_log(remote_rs_env): + """Structured request lines replace uvicorn access logs on the HTTP transport.""" + fake_app = MagicMock() + mock_http_app = MagicMock() + settings = resolve_mcp_settings( + profile="remote", transport="http", host="127.0.0.1", port=9123 + ) + with ( + patch("pipefy_mcp.server.configure_observability_logging") as mock_configure, + patch( + "pipefy_mcp.server.wire_hosted_observability", + return_value=mock_http_app, + ) as mock_wire, + patch("uvicorn.Config") as mock_config_cls, + patch("uvicorn.Server") as mock_server_cls, + ): + mock_server_cls.return_value.serve = AsyncMock() + await _serve_streamable_http(fake_app, settings) + + mock_configure.assert_called_once_with() + mock_wire.assert_called_once_with(fake_app, settings) + mock_config_cls.assert_called_once_with( + mock_http_app, + host="127.0.0.1", + port=9123, + log_level="info", + access_log=False, + ) + mock_server_cls.return_value.serve.assert_awaited_once() @pytest.mark.unit @@ -386,9 +426,12 @@ def test_run_server_http_fills_host_and_port_from_settings_when_unset( monkeypatch.delenv("PIPEFY_MCP_HOST", raising=False) monkeypatch.delenv("PIPEFY_MCP_PORT", raising=False) fake_app = MagicMock() - with patch( - "pipefy_mcp.server.build_pipefy_mcp_server", return_value=fake_app - ) as mock_build: + with ( + patch( + "pipefy_mcp.server.build_pipefy_mcp_server", return_value=fake_app + ) as mock_build, + patch("pipefy_mcp.server.anyio.run"), + ): run_server( resolve_mcp_settings( profile="remote", transport="http", host=None, port=None @@ -404,9 +447,12 @@ def test_run_server_http_fills_host_and_port_from_settings_when_unset( def test_run_server_http_respects_an_explicit_zero_port(remote_rs_env): """``port=0`` (let the OS pick) must not be swallowed as a falsy default.""" fake_app = MagicMock() - with patch( - "pipefy_mcp.server.build_pipefy_mcp_server", return_value=fake_app - ) as mock_build: + with ( + patch( + "pipefy_mcp.server.build_pipefy_mcp_server", return_value=fake_app + ) as mock_build, + patch("pipefy_mcp.server.anyio.run"), + ): run_server( resolve_mcp_settings( profile="remote", transport="http", host="127.0.0.1", port=0 @@ -426,18 +472,20 @@ def test_run_server_remote_serves_off_loopback_without_a_bind_guard(remote_rs_en refuses it (a container binds ``0.0.0.0`` and is still private). """ fake_app = MagicMock() - with patch( - "pipefy_mcp.server.build_pipefy_mcp_server", return_value=fake_app - ) as mock_build: - run_server( - resolve_mcp_settings( - profile="remote", transport="http", host="0.0.0.0", port=9123 - ) + with ( + patch( + "pipefy_mcp.server.build_pipefy_mcp_server", return_value=fake_app + ) as mock_build, + patch("pipefy_mcp.server.anyio.run") as mock_anyio_run, + ): + settings = resolve_mcp_settings( + profile="remote", transport="http", host="0.0.0.0", port=9123 ) + run_server(settings) (built_settings,), _ = mock_build.call_args assert built_settings.mcp.host == "0.0.0.0" - fake_app.run.assert_called_once_with("streamable-http") + mock_anyio_run.assert_called_once_with(_serve_streamable_http, fake_app, settings) # --- Resource-server role (OAuth 2.0 inbound bearer validation) --------------