From 8285edb6de72fe9ed1c40552b197a1fd87b5d6b6 Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Date: Tue, 7 Jul 2026 19:46:52 -0300 Subject: [PATCH 01/11] feat(mcp): add stdout JSON structured logging emitter Introduce allowlisted JSON event builders and a dedicated stdout logger with propagate=False so hosted observability coexists with FastMCP's RichHandler. Add PIPEFY_MCP_LOG_LEVEL to McpSettings and wire it through run_server and FastMCP construction. --- docs/config.md | 14 ++ .../src/pipefy_mcp/observability/__init__.py | 1 + .../pipefy_mcp/observability/json_logging.py | 138 +++++++++++ packages/mcp/src/pipefy_mcp/server.py | 3 + packages/mcp/src/pipefy_mcp/settings.py | 30 ++- packages/mcp/tests/observability/__init__.py | 0 .../tests/observability/test_json_logging.py | 221 ++++++++++++++++++ packages/mcp/tests/test_server.py | 24 ++ packages/mcp/tests/test_settings.py | 13 ++ 9 files changed, 442 insertions(+), 2 deletions(-) create mode 100644 packages/mcp/src/pipefy_mcp/observability/__init__.py create mode 100644 packages/mcp/src/pipefy_mcp/observability/json_logging.py create mode 100644 packages/mcp/tests/observability/__init__.py create mode 100644 packages/mcp/tests/observability/test_json_logging.py diff --git a/docs/config.md b/docs/config.md index c0abe5ff..6c7c8848 100644 --- a/docs/config.md +++ b/docs/config.md @@ -42,6 +42,7 @@ unified_envelope = true remote_mode = false host = "127.0.0.1" port = 8000 +log_level = "INFO" ``` Keys use **bare pydantic field names**, not the upper-case `PIPEFY_` environment variable names. The env-only aliases (`PIPEFY_TOKEN`, `PIPEFY_OAUTH_CLIENT`, ...) exist to refuse unprefixed environment leakage and do not double as TOML keys. @@ -115,3 +116,16 @@ Storing OAuth credentials in `config.toml` puts them on disk in plain text. The ``` A future file-backed keyring backend (#237) will write its credential store as `~/.config/pipefy/keyring.cfg` next to these — a separate file with its own format. + +## MCP server (`pipefy-mcp-server`) + +These variables load into `pipefy_mcp.McpSettings` (`settings.mcp`). TOML keys use the bare field names under the same top-level file (see [Schema](#schema)). + +| Variable | Default | Effect | +|----------|---------|--------| +| `PIPEFY_MCP_PROFILE` | `local` | Launch profile: `local` registers every tool; `remote` exposes only remote-safe tools and validates an inbound bearer per request. | +| `PIPEFY_MCP_TRANSPORT` | derived from profile | `stdio` or `http`. Unset: `local` → `stdio`, `remote` → `http`. | +| `PIPEFY_MCP_HOST` | `127.0.0.1` | HTTP bind host (loopback only until hosted OBO lands). | +| `PIPEFY_MCP_PORT` | `8000` | HTTP bind port. | +| `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. | diff --git a/packages/mcp/src/pipefy_mcp/observability/__init__.py b/packages/mcp/src/pipefy_mcp/observability/__init__.py new file mode 100644 index 00000000..aff7fc65 --- /dev/null +++ b/packages/mcp/src/pipefy_mcp/observability/__init__.py @@ -0,0 +1 @@ +"""Hosted observability: structured JSON request and tool-call logging.""" diff --git a/packages/mcp/src/pipefy_mcp/observability/json_logging.py b/packages/mcp/src/pipefy_mcp/observability/json_logging.py new file mode 100644 index 00000000..1f433703 --- /dev/null +++ b/packages/mcp/src/pipefy_mcp/observability/json_logging.py @@ -0,0 +1,138 @@ +"""Pure builders and stdout JSON emitter for hosted structured log events.""" + +from __future__ import annotations + +import json +import logging +import sys +from datetime import UTC, datetime +from typing import Any, Literal + +ToolCallOutcome = Literal["ok", "error"] + +OBSERVABILITY_LOGGER_NAME = "pipefy_mcp.observability.structured" + +HTTP_REQUEST_EVENT_KEYS = frozenset( + { + "event", + "timestamp", + "method", + "path", + "status", + "duration_ms", + "client_ip", + "session_id", + "request_id", + "sub", + "client_id", + } +) + +TOOL_CALL_EVENT_KEYS = frozenset( + { + "event", + "timestamp", + "tool", + "outcome", + "duration_ms", + "arg_keys", + "request_id", + } +) + + +def normalize_log_level(log_level: str) -> int: + """Map a settings/env log level string to a ``logging`` level constant.""" + normalized = log_level.upper() + try: + return getattr(logging, normalized) + except AttributeError as exc: + raise ValueError( + f"invalid log level: received {log_level!r}, expected one of " + "DEBUG, INFO, WARNING, ERROR, CRITICAL" + ) 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.""" + logger = logging.getLogger(OBSERVABILITY_LOGGER_NAME) + logger.handlers.clear() + + handler = logging.StreamHandler(sys.stdout) + handler.setFormatter(logging.Formatter("%(message)s")) + level = normalize_log_level(log_level) + handler.setLevel(level) + logger.setLevel(level) + logger.addHandler(handler) + logger.propagate = False + return logger + + +def reset_observability_logging() -> None: + """Remove observability handlers (test isolation).""" + logger = logging.getLogger(OBSERVABILITY_LOGGER_NAME) + logger.handlers.clear() + logger.propagate = True + + +def emit_structured_event(event: dict[str, Any]) -> None: + """Emit one allowlisted event as a single JSON line on stdout.""" + line = json.dumps(event, separators=(",", ":")) + logging.getLogger(OBSERVABILITY_LOGGER_NAME).info(line) + + +def _utc_timestamp_iso(timestamp: datetime | None) -> str: + instant = timestamp or datetime.now(UTC) + if instant.tzinfo is None: + instant = instant.replace(tzinfo=UTC) + return instant.astimezone(UTC).isoformat() + + +def build_http_request_event( + *, + method: str, + path: str, + status: int | None, + duration_ms: int | float, + client_ip: str | None, + session_id: str | None, + request_id: str, + sub: str | None, + client_id: str | None, + timestamp: datetime | None = None, +) -> dict[str, Any]: + """Build the allowlisted dict for one HTTP request log line.""" + return { + "event": "http_request", + "timestamp": _utc_timestamp_iso(timestamp), + "method": method, + "path": path, + "status": status, + "duration_ms": duration_ms, + "client_ip": client_ip, + "session_id": session_id, + "request_id": request_id, + "sub": sub, + "client_id": client_id, + } + + +def build_tool_call_event( + *, + tool: str, + outcome: ToolCallOutcome, + duration_ms: int | float, + arg_keys: list[str], + request_id: str | None, + timestamp: datetime | None = None, +) -> dict[str, Any]: + """Build the allowlisted dict for one tool-call log line.""" + return { + "event": "tool_call", + "timestamp": _utc_timestamp_iso(timestamp), + "tool": tool, + "outcome": outcome, + "duration_ms": duration_ms, + "arg_keys": arg_keys, + "request_id": request_id, + } diff --git a/packages/mcp/src/pipefy_mcp/server.py b/packages/mcp/src/pipefy_mcp/server.py index fe9095a7..c04cd9ad 100644 --- a/packages/mcp/src/pipefy_mcp/server.py +++ b/packages/mcp/src/pipefy_mcp/server.py @@ -8,6 +8,7 @@ from mcp.server.fastmcp import FastMCP from pipefy_mcp.core.runtime import McpRuntime +from pipefy_mcp.observability.json_logging import configure_observability_logging from pipefy_mcp.settings import Settings from pipefy_mcp.tools.registry import ToolRegistry from pipefy_mcp.tools.validation_envelope import install_pipefy_validation_envelope @@ -97,6 +98,7 @@ def build_pipefy_mcp_server(settings: Settings) -> FastMCP: lifespan=_make_lifespan(runtime), host=settings.mcp.host, port=settings.mcp.port, + log_level=settings.mcp.log_level, token_verifier=verifier, auth=auth, ) @@ -156,6 +158,7 @@ def run_server(settings: Settings) -> None: HTTP's bind concerns. """ 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) diff --git a/packages/mcp/src/pipefy_mcp/settings.py b/packages/mcp/src/pipefy_mcp/settings.py index 13945d72..a6756bdd 100644 --- a/packages/mcp/src/pipefy_mcp/settings.py +++ b/packages/mcp/src/pipefy_mcp/settings.py @@ -6,13 +6,21 @@ from pipefy_infra import security from pipefy_infra.config import PipefyTomlConfigSource from pipefy_sdk import PipefySettings -from pydantic import AliasChoices, Field, ValidationError, model_validator +from pydantic import ( + AliasChoices, + Field, + ValidationError, + field_validator, + model_validator, +) from pydantic_settings import ( BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict, ) +McpLogLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] + class McpSettings(BaseSettings): """MCP-server runtime knobs: transport, tool exposure, and envelope shape. @@ -23,7 +31,7 @@ class McpSettings(BaseSettings): re-attaches it so the operator-facing ``PIPEFY_MCP_*`` env vars stay byte-identical. The shared ``config.toml`` source keys off the bare field names, so TOML keys are ``unified_envelope``, ``profile``, ``transport``, - ``host``, ``port``. + ``host``, ``port``, ``log_level``. """ model_config = SettingsConfigDict( @@ -102,6 +110,23 @@ 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). " + "Accepts DEBUG, INFO, WARNING, ERROR, or CRITICAL; normalized to " + "uppercase." + ), + ) + + @field_validator("log_level", mode="before") + @classmethod + def _normalize_log_level(cls, value: object) -> object: + if isinstance(value, str): + return value.upper() + return value + @model_validator(mode="after") def _resolve_transport(self) -> Self: """Fill the profile-derived transport default and reject 'remote' over stdio. @@ -279,6 +304,7 @@ def resolve_mcp_settings( __all__ = [ "AuthSettings", "JwtValidationSettings", + "McpLogLevel", "McpSettings", "PipefySettings", "ResourceServerSettings", diff --git a/packages/mcp/tests/observability/__init__.py b/packages/mcp/tests/observability/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/packages/mcp/tests/observability/test_json_logging.py b/packages/mcp/tests/observability/test_json_logging.py new file mode 100644 index 00000000..bc574866 --- /dev/null +++ b/packages/mcp/tests/observability/test_json_logging.py @@ -0,0 +1,221 @@ +"""Tests for structured JSON log event builders.""" + +from __future__ import annotations + +import json +import logging +import sys +from datetime import UTC, datetime + +import pytest + +from pipefy_mcp.observability.json_logging import ( + HTTP_REQUEST_EVENT_KEYS, + OBSERVABILITY_LOGGER_NAME, + TOOL_CALL_EVENT_KEYS, + ToolCallOutcome, + build_http_request_event, + build_tool_call_event, + configure_observability_logging, + emit_structured_event, + reset_observability_logging, +) + +_FIXED_TIMESTAMP = datetime(2026, 7, 7, 22, 15, 30, tzinfo=UTC) + + +class TestBuildHttpRequestEvent: + def test_has_expected_fields(self): + event = build_http_request_event( + method="POST", + path="/mcp", + status=200, + duration_ms=42, + client_ip="127.0.0.1", + session_id="sess-abc", + request_id="req-123", + sub="user-subject", + client_id="client-azp", + timestamp=_FIXED_TIMESTAMP, + ) + + assert set(event.keys()) == HTTP_REQUEST_EVENT_KEYS + assert event == { + "event": "http_request", + "timestamp": "2026-07-07T22:15:30+00:00", + "method": "POST", + "path": "/mcp", + "status": 200, + "duration_ms": 42, + "client_ip": "127.0.0.1", + "session_id": "sess-abc", + "request_id": "req-123", + "sub": "user-subject", + "client_id": "client-azp", + } + + def test_null_identity_fields_when_absent(self): + event = build_http_request_event( + method="GET", + path="/health", + status=None, + duration_ms=1, + client_ip=None, + session_id=None, + request_id="req-null", + sub=None, + client_id=None, + timestamp=_FIXED_TIMESTAMP, + ) + + assert event["sub"] is None + assert event["client_id"] is None + assert event["status"] is None + + def test_output_is_json_serializable_one_line(self): + event = build_http_request_event( + method="POST", + path="/mcp", + status=200, + duration_ms=10, + client_ip="127.0.0.1", + session_id="s", + request_id="r", + sub=None, + client_id=None, + timestamp=_FIXED_TIMESTAMP, + ) + + line = json.dumps(event, separators=(",", ":")) + assert "\n" not in line + assert json.loads(line) == event + + +class TestBuildToolCallEvent: + def test_has_expected_fields(self): + event = build_tool_call_event( + tool="get_card", + outcome="ok", + duration_ms=15, + arg_keys=["card_id"], + request_id="req-456", + timestamp=_FIXED_TIMESTAMP, + ) + + assert set(event.keys()) == TOOL_CALL_EVENT_KEYS + assert event == { + "event": "tool_call", + "timestamp": "2026-07-07T22:15:30+00:00", + "tool": "get_card", + "outcome": "ok", + "duration_ms": 15, + "arg_keys": ["card_id"], + "request_id": "req-456", + } + + @pytest.mark.parametrize("outcome", ["ok", "error"]) + def test_outcome_is_ok_or_error(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", + timestamp=_FIXED_TIMESTAMP, + ) + + assert event["outcome"] == outcome + + def test_request_id_null_when_uncorrelated(self): + event = build_tool_call_event( + tool="get_pipe", + outcome="ok", + duration_ms=2, + arg_keys=[], + request_id=None, + timestamp=_FIXED_TIMESTAMP, + ) + + assert event["request_id"] is None + + +class TestObservabilityLoggingEmitter: + @pytest.fixture(autouse=True) + def _isolated_logger(self): + reset_observability_logging() + yield + reset_observability_logging() + + def test_emits_valid_json_one_liner_on_stdout(self, capsys): + configure_observability_logging(log_level="INFO") + event = build_http_request_event( + method="POST", + path="/mcp", + status=200, + duration_ms=10, + client_ip="127.0.0.1", + session_id=None, + request_id="req-emit", + sub=None, + client_id=None, + timestamp=_FIXED_TIMESTAMP, + ) + + emit_structured_event(event) + + line = capsys.readouterr().out.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, + ) + ) + + assert capsys.readouterr().out == "" + + def test_does_not_propagate_to_root_logger(self): + root = logging.getLogger() + captured: list[logging.LogRecord] = [] + + class _CapturingHandler(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + captured.append(record) + + root_handler = _CapturingHandler() + root.addHandler(root_handler) + root.setLevel(logging.DEBUG) + try: + configure_observability_logging(log_level="INFO") + emit_structured_event( + build_http_request_event( + method="GET", + path="/mcp", + status=200, + duration_ms=1, + client_ip=None, + session_id=None, + request_id="req-root", + sub=None, + client_id=None, + timestamp=_FIXED_TIMESTAMP, + ) + ) + assert captured == [] + finally: + root.removeHandler(root_handler) + + def test_handler_targets_stdout_explicitly(self): + configure_observability_logging(log_level="INFO") + logger = logging.getLogger(OBSERVABILITY_LOGGER_NAME) + assert len(logger.handlers) == 1 + assert logger.handlers[0].stream is sys.stdout + assert logger.propagate is False diff --git a/packages/mcp/tests/test_server.py b/packages/mcp/tests/test_server.py index f3cc6848..21703828 100644 --- a/packages/mcp/tests/test_server.py +++ b/packages/mcp/tests/test_server.py @@ -93,6 +93,30 @@ async def test_register_tools(mocked_runtime): ) +@pytest.mark.unit +def test_build_pipefy_mcp_server_passes_log_level_to_fastmcp(mocked_runtime): + settings = _MINIMAL_PIPEFY_SETTINGS.model_copy( + update={"mcp": McpSettings(log_level="WARNING")} + ) + with patch("pipefy_mcp.server.FastMCP") as mock_fastmcp: + build_pipefy_mcp_server(settings) + assert mock_fastmcp.call_args.kwargs["log_level"] == "WARNING" + + +@pytest.mark.unit +def test_run_server_configures_structured_logging(monkeypatch): + monkeypatch.delenv("PIPEFY_MCP_PROFILE", raising=False) + monkeypatch.delenv("PIPEFY_MCP_TRANSPORT", raising=False) + with ( + patch("pipefy_mcp.server.configure_observability_logging") as mock_configure, + patch("pipefy_mcp.server.build_pipefy_mcp_server"), + ): + run_server( + resolve_mcp_settings(profile=None, transport=None, host=None, port=None) + ) + mock_configure.assert_called_once_with(log_level="INFO") + + @pytest.mark.unit def test_run_server_builds_the_stdio_server_and_runs_it(monkeypatch): """The default (local/stdio) profile builds at startup and delegates to mcp.run().""" diff --git a/packages/mcp/tests/test_settings.py b/packages/mcp/tests/test_settings.py index cbc21aae..126ac4c0 100644 --- a/packages/mcp/tests/test_settings.py +++ b/packages/mcp/tests/test_settings.py @@ -163,6 +163,19 @@ def test_mcp_settings_defaults(): assert mcp.transport == "stdio" assert mcp.host == "127.0.0.1" assert mcp.port == 8000 + assert mcp.log_level == "INFO" + + +@pytest.mark.unit +def test_mcp_settings_log_level_from_env_normalizes_case(monkeypatch): + monkeypatch.setenv("PIPEFY_MCP_LOG_LEVEL", "warning") + assert Settings().mcp.log_level == "WARNING" + + +@pytest.mark.unit +def test_mcp_settings_log_level_rejects_unknown_value(): + with pytest.raises(ValidationError): + McpSettings(log_level="verbose") @pytest.mark.unit From 9d0f061244352429187949a07ca74615155e3346 Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Date: Tue, 7 Jul 2026 19:47:03 -0300 Subject: [PATCH 02/11] feat(mcp): preserve JWT sub claim in PipefyAccessToken Extend the resource-server token mapping with an optional sub field so hosted HTTP request logs can emit caller subject alongside client_id without re-decoding the bearer. --- .../src/pipefy_mcp/auth/resource_server.py | 9 +++++++- .../mcp/tests/auth/test_resource_server.py | 21 +++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/packages/mcp/src/pipefy_mcp/auth/resource_server.py b/packages/mcp/src/pipefy_mcp/auth/resource_server.py index 8b80e45c..d9640d3d 100644 --- a/packages/mcp/src/pipefy_mcp/auth/resource_server.py +++ b/packages/mcp/src/pipefy_mcp/auth/resource_server.py @@ -39,6 +39,12 @@ logger = logging.getLogger(__name__) +class PipefyAccessToken(AccessToken): + """AccessToken with the JWT ``sub`` claim preserved for request logging.""" + + sub: str | None = None + + def _parse_scopes(scope: Any) -> list[str]: """Normalize the ``scope`` claim to a list of strings. @@ -104,7 +110,7 @@ def _to_access_token(self, token: str, claims: dict[str, Any]) -> AccessToken: if exp is None: raise ValueError("token has no exp claim") - return AccessToken( + return PipefyAccessToken( token=token, client_id=client_id, scopes=_parse_scopes(claims.get("scope")), @@ -112,6 +118,7 @@ def _to_access_token(self, token: str, claims: dict[str, Any]) -> AccessToken: # wants an int. expires_at=int(exp), resource=self._resource, + sub=claims.get("sub"), ) diff --git a/packages/mcp/tests/auth/test_resource_server.py b/packages/mcp/tests/auth/test_resource_server.py index 9f0ab01c..4171687f 100644 --- a/packages/mcp/tests/auth/test_resource_server.py +++ b/packages/mcp/tests/auth/test_resource_server.py @@ -50,11 +50,32 @@ async def test_maps_claims_to_access_token() -> None: assert token is not None assert token.token == "the-token" assert token.client_id == "client-abc" + assert token.sub == "user-123" assert token.scopes == ["read", "write"] assert token.expires_at == _EXP assert token.resource == _RESOURCE +@pytest.mark.unit +async def test_sub_is_none_when_claim_absent() -> None: + token = await JwtTokenVerifier( + _StubValidator(claims={"azp": "client-abc", "exp": _EXP}) + ).verify_token("t") + assert token is not None + assert token.client_id == "client-abc" + assert token.sub is None + + +@pytest.mark.unit +async def test_sub_preserved_when_azp_is_client_id() -> None: + token = await JwtTokenVerifier( + _StubValidator(claims={"azp": "client-abc", "sub": "user-123", "exp": _EXP}) + ).verify_token("t") + assert token is not None + assert token.client_id == "client-abc" + assert token.sub == "user-123" + + @pytest.mark.unit async def test_client_id_falls_back_to_client_id_then_sub() -> None: by_client_id = await JwtTokenVerifier( From 33662e014f77a21097248df087c6b36e4e16283c Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Date: Tue, 7 Jul 2026 20:19:58 -0300 Subject: [PATCH 03/11] fix(mcp): map log levels explicitly and re-export the emitter surface normalize_log_level resolved names via getattr(logging, ...), which returns non-level module attributes (BASIC_FORMAT) and accepts aliases the settings Literal rejects (WARN, FATAL); an explicit map fails loudly on both. The observability package now re-exports its stdlib-only emitter surface per the package convention, and a suite-wide autouse fixture drops leftover stdout handlers so a test that configures the process-global logger cannot leak a closed-stream handler into later tests. --- .../src/pipefy_mcp/observability/__init__.py | 27 ++++++++++++++++- .../pipefy_mcp/observability/json_logging.py | 20 +++++++++++-- packages/mcp/tests/conftest.py | 15 ++++++++++ .../tests/observability/test_json_logging.py | 30 +++++++++++++++++++ 4 files changed, 88 insertions(+), 4 deletions(-) diff --git a/packages/mcp/src/pipefy_mcp/observability/__init__.py b/packages/mcp/src/pipefy_mcp/observability/__init__.py index aff7fc65..f4020862 100644 --- a/packages/mcp/src/pipefy_mcp/observability/__init__.py +++ b/packages/mcp/src/pipefy_mcp/observability/__init__.py @@ -1 +1,26 @@ -"""Hosted observability: structured JSON request and tool-call logging.""" +"""Hosted observability: structured JSON request and tool-call logging. + +Only the stdlib-only emitter surface is re-exported here; the middleware and +wiring stay submodule imports so importing this package never pulls +``starlette`` or the MCP SDK. +""" + +from pipefy_mcp.observability.json_logging import ( + OBSERVABILITY_LOGGER_NAME, + build_http_request_event, + build_tool_call_event, + configure_observability_logging, + emit_structured_event, + normalize_log_level, + reset_observability_logging, +) + +__all__ = [ + "OBSERVABILITY_LOGGER_NAME", + "build_http_request_event", + "build_tool_call_event", + "configure_observability_logging", + "emit_structured_event", + "normalize_log_level", + "reset_observability_logging", +] diff --git a/packages/mcp/src/pipefy_mcp/observability/json_logging.py b/packages/mcp/src/pipefy_mcp/observability/json_logging.py index 1f433703..a33fbcf6 100644 --- a/packages/mcp/src/pipefy_mcp/observability/json_logging.py +++ b/packages/mcp/src/pipefy_mcp/observability/json_logging.py @@ -41,12 +41,26 @@ ) +_LOG_LEVELS = { + "DEBUG": logging.DEBUG, + "INFO": logging.INFO, + "WARNING": logging.WARNING, + "ERROR": logging.ERROR, + "CRITICAL": logging.CRITICAL, +} + + def normalize_log_level(log_level: str) -> int: - """Map a settings/env log level string to a ``logging`` level constant.""" + """Map a settings/env log level string to a ``logging`` level constant. + + An explicit map rather than ``getattr(logging, ...)``: the module carries + uppercase attributes that are not levels (``BASIC_FORMAT``) and aliases the + settings ``Literal`` rejects (``WARN``, ``FATAL``); both must fail here too. + """ normalized = log_level.upper() try: - return getattr(logging, normalized) - except AttributeError as exc: + return _LOG_LEVELS[normalized] + except KeyError as exc: raise ValueError( f"invalid log level: received {log_level!r}, expected one of " "DEBUG, INFO, WARNING, ERROR, CRITICAL" diff --git a/packages/mcp/tests/conftest.py b/packages/mcp/tests/conftest.py index ea8651d5..b1ff4677 100644 --- a/packages/mcp/tests/conftest.py +++ b/packages/mcp/tests/conftest.py @@ -2,6 +2,7 @@ import pytest +from pipefy_mcp.observability.json_logging import reset_observability_logging from pipefy_mcp.tools.validation_envelope import install_pipefy_validation_envelope # Same validation envelope as ``server.py`` lifespan (idempotent). @@ -30,3 +31,17 @@ def clear_auth_env(monkeypatch): """Strip ambient ``PIPEFY_*`` auth env so ``AuthSettings()`` is hermetic.""" for key in _AUTH_ENV_KEYS: monkeypatch.delenv(key, raising=False) + + +@pytest.fixture(autouse=True) +def _reset_observability_logging_between_tests(): + """Drop observability handlers after every test. + + 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 + later ``emit_structured_event`` in an unrelated test would write into a + closed stream. + """ + yield + reset_observability_logging() diff --git a/packages/mcp/tests/observability/test_json_logging.py b/packages/mcp/tests/observability/test_json_logging.py index bc574866..91e76107 100644 --- a/packages/mcp/tests/observability/test_json_logging.py +++ b/packages/mcp/tests/observability/test_json_logging.py @@ -18,6 +18,7 @@ build_tool_call_event, configure_observability_logging, emit_structured_event, + normalize_log_level, reset_observability_logging, ) @@ -219,3 +220,32 @@ def test_handler_targets_stdout_explicitly(self): assert len(logger.handlers) == 1 assert logger.handlers[0].stream is sys.stdout 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") + + emit_structured_event( + build_tool_call_event( + tool="get_card", + outcome="ok", + duration_ms=1, + arg_keys=[], + request_id="req-once", + timestamp=_FIXED_TIMESTAMP, + ) + ) + + logger = logging.getLogger(OBSERVABILITY_LOGGER_NAME) + assert len(logger.handlers) == 1 + assert len(capsys.readouterr().out.strip().splitlines()) == 1 + + def test_normalize_log_level_rejects_unknown_name(self): + with pytest.raises(ValueError, match="invalid log level"): + normalize_log_level("verbose") + + def test_normalize_log_level_rejects_non_level_logging_attribute(self): + # getattr(logging, ...) resolves module attributes that are not levels; + # anything that does not map to an int level must be rejected. + with pytest.raises(ValueError, match="invalid log level"): + normalize_log_level("BASIC_FORMAT") From 932c9f2c1bb19abecbc54988d10f22bad5790fbb Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Date: Tue, 7 Jul 2026 20:20:12 -0300 Subject: [PATCH 04/11] fix(mcp): degrade a non-string JWT sub to None instead of rejecting the token PipefyAccessToken.sub is str | None, so a validly-signed token carrying a numeric sub raised ValidationError inside _to_access_token and verify_token turned it into a 401, a token dev accepted before the field existed. sub feeds request logging only; a malformed value must not gate authentication. RFC 7519 requires StringOrURI, so only non-compliant IdPs ever hit this. --- packages/mcp/src/pipefy_mcp/auth/resource_server.py | 7 ++++++- packages/mcp/tests/auth/test_resource_server.py | 11 +++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/packages/mcp/src/pipefy_mcp/auth/resource_server.py b/packages/mcp/src/pipefy_mcp/auth/resource_server.py index d9640d3d..14c307ca 100644 --- a/packages/mcp/src/pipefy_mcp/auth/resource_server.py +++ b/packages/mcp/src/pipefy_mcp/auth/resource_server.py @@ -110,6 +110,11 @@ def _to_access_token(self, token: str, claims: dict[str, Any]) -> AccessToken: if exp is None: raise ValueError("token has no exp claim") + # sub feeds request logging only; a malformed (non-string) sub must + # degrade to None, not reject a token the verifier accepted before the + # field existed. + sub_claim = claims.get("sub") + return PipefyAccessToken( token=token, client_id=client_id, @@ -118,7 +123,7 @@ def _to_access_token(self, token: str, claims: dict[str, Any]) -> AccessToken: # wants an int. expires_at=int(exp), resource=self._resource, - sub=claims.get("sub"), + sub=sub_claim if isinstance(sub_claim, str) else None, ) diff --git a/packages/mcp/tests/auth/test_resource_server.py b/packages/mcp/tests/auth/test_resource_server.py index 4171687f..23ba1e23 100644 --- a/packages/mcp/tests/auth/test_resource_server.py +++ b/packages/mcp/tests/auth/test_resource_server.py @@ -76,6 +76,17 @@ async def test_sub_preserved_when_azp_is_client_id() -> None: assert token.sub == "user-123" +@pytest.mark.unit +async def test_non_string_sub_degrades_to_none_instead_of_rejecting() -> None: + """sub feeds logging only; a malformed sub must not reject a valid token.""" + token = await JwtTokenVerifier( + _StubValidator(claims={"azp": "client-abc", "sub": 12345, "exp": _EXP}) + ).verify_token("t") + assert token is not None + assert token.client_id == "client-abc" + assert token.sub is None + + @pytest.mark.unit async def test_client_id_falls_back_to_client_id_then_sub() -> None: by_client_id = await JwtTokenVerifier( From a03af02ea9536f5a3c7fcd86bda86a97fb47f6ef Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Date: Tue, 7 Jul 2026 19:47:03 -0300 Subject: [PATCH 05/11] feat(mcp): add pure-ASGI HTTP request log middleware Emit one structured JSON line per HTTP request from middleware that only inspects response.start, stores request_id in scope state, and reads caller identity from the auth context at emit time. --- .../observability/request_log_middleware.py | 117 +++++++ .../test_request_log_middleware.py | 310 ++++++++++++++++++ 2 files changed, 427 insertions(+) create mode 100644 packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py create mode 100644 packages/mcp/tests/observability/test_request_log_middleware.py 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..11c74880 --- /dev/null +++ b/packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py @@ -0,0 +1,117 @@ +"""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.auth_context 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" + + +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 _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 = str(uuid4()) + 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/tests/observability/test_request_log_middleware.py b/packages/mcp/tests/observability/test_request_log_middleware.py new file mode 100644 index 00000000..beb78e57 --- /dev/null +++ b/packages/mcp/tests/observability/test_request_log_middleware.py @@ -0,0 +1,310 @@ +"""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.auth_context import AuthenticatedUser +from mcp.server.auth.provider import AccessToken + +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 + +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().out.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(log_level="INFO") + + +@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) + 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_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_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)) + await _run_middleware(app, scope) + + lines = _read_log_lines(capsys) + assert lines[0]["sub"] == "user-123" + assert lines[0]["client_id"] == "client-abc" + + +@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)) + await _run_middleware(app, scope) + + lines = _read_log_lines(capsys) + assert lines[0]["sub"] is None + assert lines[0]["client_id"] == "client-abc" From 8f5e57fed78517d31302550741633c014e8d184c Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Date: Tue, 7 Jul 2026 19:47:03 -0300 Subject: [PATCH 06/11] feat(mcp): serve HTTP with request logging via manual uvicorn Wire RequestLogMiddleware through streamable_http_app once and replace FastMCP.run(streamable-http) with explicit uvicorn serve and access_log disabled so structured request lines replace text access logs. --- packages/mcp/AGENTS.md | 17 +++ .../src/pipefy_mcp/observability/wiring.py | 21 ++++ packages/mcp/src/pipefy_mcp/server.py | 21 +++- .../mcp/tests/observability/test_wiring.py | 43 +++++++ packages/mcp/tests/test_server.py | 105 ++++++++++++------ 5 files changed, 175 insertions(+), 32 deletions(-) create mode 100644 packages/mcp/src/pipefy_mcp/observability/wiring.py create mode 100644 packages/mcp/tests/observability/test_wiring.py diff --git a/packages/mcp/AGENTS.md b/packages/mcp/AGENTS.md index b5a590d3..1b6a4032 100644 --- a/packages/mcp/AGENTS.md +++ b/packages/mcp/AGENTS.md @@ -80,6 +80,23 @@ stays off until that lands (see `experiments/hosted-obo/RFC-OUTLINE.md`). The attachment tools' local `file_path` 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 one JSON line per request on stdout for hosted +observability (`pipefy_mcp/observability/`). Stdio does **not** install request +logging: under stdio, stdout is the JSON-RPC wire, so structured lines there +would corrupt the protocol. + +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. + +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. + ## Tool registration Tools are registered **once, at construction** (via `_register_pipefy_tools` in 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 9c23d579..b5ef8913 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 @@ -166,6 +168,23 @@ def _assert_safe_http_bind(*, host: str) -> None: ) +async def _serve_streamable_http(app: FastMCP, settings: Settings) -> None: + """Serve Streamable HTTP with hosted observability middleware wired in.""" + import uvicorn + + 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. @@ -216,4 +235,4 @@ def run_server(settings: Settings) -> None: ) _assert_safe_http_bind(host=mcp.host) - build_pipefy_mcp_server(settings).run("streamable-http") + anyio.run(_serve_streamable_http, build_pipefy_mcp_server(settings), settings) diff --git a/packages/mcp/tests/observability/test_wiring.py b/packages/mcp/tests/observability/test_wiring.py new file mode 100644 index 00000000..f74cd712 --- /dev/null +++ b/packages/mcp/tests/observability/test_wiring.py @@ -0,0 +1,43 @@ +"""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 95196076..ae705f89 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 @@ -18,6 +18,7 @@ _assert_safe_http_bind, _make_lifespan, _register_pipefy_tools, + _serve_streamable_http, build_pipefy_mcp_server, default_tool_middlewares, run_server, @@ -317,21 +318,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 @@ -341,16 +344,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" @@ -359,20 +364,52 @@ 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.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_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 @@ -403,9 +440,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 @@ -421,9 +461,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 From 44ce147f01e1e57e80180944f22842e45e4bc3b1 Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Date: Tue, 7 Jul 2026 20:20:12 -0300 Subject: [PATCH 07/11] fix(mcp): configure the stdout emitter only on the HTTP serve path configure_observability_logging ran before the transport branch, arming a stdout handler in the stdio process too, where stdout is the JSON-RPC wire. Nothing emits under stdio today, but the D10 guarantee was conventional rather than structural: one future emit call in shared code would corrupt the protocol. The call now lives in _serve_streamable_http, and the stdio test asserts configuration never happens there. Also imports AuthenticatedUser from its defining module (bearer_auth) and extends the middleware tests: bearer absent from output with an Authorization header planted, UnauthenticatedUser yields null identity, the ordering test fails fast instead of hanging, and a send that raises mid-stream still emits exactly one line. --- .../observability/request_log_middleware.py | 2 +- packages/mcp/src/pipefy_mcp/server.py | 10 ++- .../test_request_log_middleware.py | 80 ++++++++++++++++--- packages/mcp/tests/test_server.py | 7 +- 4 files changed, 81 insertions(+), 18 deletions(-) diff --git a/packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py b/packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py index 11c74880..dde65bca 100644 --- a/packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py +++ b/packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py @@ -7,7 +7,7 @@ from typing import Any from uuid import uuid4 -from mcp.server.auth.middleware.auth_context import AuthenticatedUser +from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser from pipefy_mcp.auth.resource_server import PipefyAccessToken from pipefy_mcp.observability.json_logging import ( diff --git a/packages/mcp/src/pipefy_mcp/server.py b/packages/mcp/src/pipefy_mcp/server.py index b5ef8913..2dda51c3 100644 --- a/packages/mcp/src/pipefy_mcp/server.py +++ b/packages/mcp/src/pipefy_mcp/server.py @@ -169,9 +169,16 @@ def _assert_safe_http_bind(*, host: str) -> None: async def _serve_streamable_http(app: FastMCP, settings: Settings) -> None: - """Serve Streamable HTTP with hosted observability middleware wired in.""" + """Serve Streamable HTTP with hosted observability middleware wired in. + + The stdout emitter is configured here, not in :func:`run_server`, so the + stdio path never carries a stdout log handler: under stdio, stdout is the + JSON-RPC wire, and a configured handler there would be one stray + ``emit_structured_event`` call away from corrupting the protocol. + """ import uvicorn + configure_observability_logging(log_level=settings.mcp.log_level) http_app = wire_hosted_observability(app, settings) mcp = settings.mcp config = uvicorn.Config( @@ -214,7 +221,6 @@ def run_server(settings: Settings) -> None: HTTP's bind concerns. """ 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) diff --git a/packages/mcp/tests/observability/test_request_log_middleware.py b/packages/mcp/tests/observability/test_request_log_middleware.py index beb78e57..e13cea73 100644 --- a/packages/mcp/tests/observability/test_request_log_middleware.py +++ b/packages/mcp/tests/observability/test_request_log_middleware.py @@ -8,8 +8,9 @@ import anyio import pytest -from mcp.server.auth.middleware.auth_context import AuthenticatedUser +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 ( @@ -148,16 +149,17 @@ async def recording_send(message: dict[str, Any]) -> None: downstream_messages.append(message) middleware = RequestLogMiddleware(streaming_app) - 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() + 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 @@ -188,6 +190,32 @@ async def disconnect_receive() -> dict[str, Any]: 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() @@ -248,12 +276,34 @@ 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)) + 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 @@ -302,9 +352,13 @@ 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)) + 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/test_server.py b/packages/mcp/tests/test_server.py index ae705f89..8a600260 100644 --- a/packages/mcp/tests/test_server.py +++ b/packages/mcp/tests/test_server.py @@ -109,7 +109,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 carry a stdout log handler: stdout is the JSON-RPC wire.""" monkeypatch.delenv("PIPEFY_MCP_PROFILE", raising=False) monkeypatch.delenv("PIPEFY_MCP_TRANSPORT", raising=False) with ( @@ -119,7 +120,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 @@ -391,6 +392,7 @@ async def test_serve_streamable_http_disables_uvicorn_access_log(remote_rs_env): 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, @@ -401,6 +403,7 @@ async def test_serve_streamable_http_disables_uvicorn_access_log(remote_rs_env): mock_server_cls.return_value.serve = AsyncMock() await _serve_streamable_http(fake_app, settings) + mock_configure.assert_called_once_with(log_level="INFO") mock_wire.assert_called_once_with(fake_app, settings) mock_config_cls.assert_called_once_with( mock_http_app, From ed0eec8419b9eddeba3533b06ca4e41178b780fe Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Date: Thu, 9 Jul 2026 14:34:33 -0300 Subject: [PATCH 08/11] fix(mcp): emit structured JSON logs on stderr Route the hosted observability handler to stderr so structured lines never share the stdio JSON-RPC stdout channel. Collectors already capture both streams; HTTP-only configure remains the gate that keeps local installs clean. --- docs/config.md | 2 +- packages/mcp/AGENTS.md | 6 +++--- .../src/pipefy_mcp/observability/json_logging.py | 8 ++++---- packages/mcp/src/pipefy_mcp/server.py | 8 ++++---- packages/mcp/src/pipefy_mcp/settings.py | 2 +- packages/mcp/tests/conftest.py | 2 +- .../mcp/tests/observability/test_json_logging.py | 16 ++++++++++------ .../observability/test_request_log_middleware.py | 2 +- packages/mcp/tests/observability/test_wiring.py | 4 +++- packages/mcp/tests/test_server.py | 2 +- 10 files changed, 29 insertions(+), 23 deletions(-) diff --git a/docs/config.md b/docs/config.md index 6c7c8848..aaa82473 100644 --- a/docs/config.md +++ b/docs/config.md @@ -128,4 +128,4 @@ These variables load into `pipefy_mcp.McpSettings` (`settings.mcp`). TOML keys u | `PIPEFY_MCP_HOST` | `127.0.0.1` | HTTP bind host (loopback only until hosted OBO lands). | | `PIPEFY_MCP_PORT` | `8000` | HTTP bind port. | | `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 **both** hosted structured JSON lines on **stderr** 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. Structured lines use stderr so they never share the stdio JSON-RPC stdout channel. | diff --git a/packages/mcp/AGENTS.md b/packages/mcp/AGENTS.md index 1b6a4032..893258a2 100644 --- a/packages/mcp/AGENTS.md +++ b/packages/mcp/AGENTS.md @@ -82,10 +82,10 @@ shares the client's disk (remote-safe file inputs are separate follow-up work). ## Hosted structured logging -The HTTP transport emits one JSON line per request on stdout for hosted +The HTTP transport emits one JSON line per request on stderr for hosted observability (`pipefy_mcp/observability/`). Stdio does **not** install request -logging: under stdio, stdout is the JSON-RPC wire, so structured lines there -would corrupt the protocol. +logging: under stdio, stdout is the JSON-RPC wire, and the structured emitter is +HTTP-only so local installs never 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/json_logging.py b/packages/mcp/src/pipefy_mcp/observability/json_logging.py index a33fbcf6..fd7bf653 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 @@ -68,11 +68,11 @@ def normalize_log_level(log_level: str) -> int: def configure_observability_logging(*, log_level: str) -> logging.Logger: - """Attach a stdout JSON-line handler that does not propagate to the root logger.""" + """Attach a stderr JSON-line handler that does not propagate to the root logger.""" 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) @@ -90,7 +90,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) diff --git a/packages/mcp/src/pipefy_mcp/server.py b/packages/mcp/src/pipefy_mcp/server.py index 2dda51c3..6be2a9d9 100644 --- a/packages/mcp/src/pipefy_mcp/server.py +++ b/packages/mcp/src/pipefy_mcp/server.py @@ -171,10 +171,10 @@ def _assert_safe_http_bind(*, host: str) -> None: async def _serve_streamable_http(app: FastMCP, settings: Settings) -> None: """Serve Streamable HTTP with hosted observability middleware wired in. - The stdout emitter is configured here, not in :func:`run_server`, so the - stdio path never carries a stdout log handler: under stdio, stdout is the - JSON-RPC wire, and a configured handler there would be one stray - ``emit_structured_event`` call away from corrupting the protocol. + 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 diff --git a/packages/mcp/src/pipefy_mcp/settings.py b/packages/mcp/src/pipefy_mcp/settings.py index a6756bdd..dadfce9f 100644 --- a/packages/mcp/src/pipefy_mcp/settings.py +++ b/packages/mcp/src/pipefy_mcp/settings.py @@ -113,7 +113,7 @@ def settings_customise_sources( log_level: McpLogLevel = Field( default="INFO", description=( - "Log level for hosted structured JSON events on stdout and the " + "Log level for hosted structured JSON events on stderr and the " "FastMCP root logger on stderr (env: PIPEFY_MCP_LOG_LEVEL). " "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..5494edfd 100644 --- a/packages/mcp/tests/observability/test_json_logging.py +++ b/packages/mcp/tests/observability/test_json_logging.py @@ -147,7 +147,7 @@ def _isolated_logger(self): yield reset_observability_logging() - def test_emits_valid_json_one_liner_on_stdout(self, capsys): + def test_emits_valid_json_one_liner_on_stderr(self, capsys): configure_observability_logging(log_level="INFO") event = build_http_request_event( method="POST", @@ -164,7 +164,9 @@ 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 @@ -181,7 +183,9 @@ def test_warning_level_suppresses_info_events(self, capsys): ) ) - assert capsys.readouterr().out == "" + captured = capsys.readouterr() + assert captured.out == "" + assert captured.err == "" def test_does_not_propagate_to_root_logger(self): root = logging.getLogger() @@ -214,11 +218,11 @@ def emit(self, record: logging.LogRecord) -> None: finally: root.removeHandler(root_handler) - def test_handler_targets_stdout_explicitly(self): + def test_handler_targets_stderr_explicitly(self): configure_observability_logging(log_level="INFO") 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.propagate is False def test_configure_twice_keeps_one_handler_and_one_line(self, capsys): @@ -238,7 +242,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 index e13cea73..cadb8eaa 100644 --- a/packages/mcp/tests/observability/test_request_log_middleware.py +++ b/packages/mcp/tests/observability/test_request_log_middleware.py @@ -72,7 +72,7 @@ async def recording_send(message: dict[str, Any]) -> None: def _read_log_lines(capsys: pytest.CaptureFixture[str]) -> list[dict[str, Any]]: return [ json.loads(line) - for line in capsys.readouterr().out.splitlines() + for line in capsys.readouterr().err.splitlines() if line.strip() ] diff --git a/packages/mcp/tests/observability/test_wiring.py b/packages/mcp/tests/observability/test_wiring.py index f74cd712..eb4b5f81 100644 --- a/packages/mcp/tests/observability/test_wiring.py +++ b/packages/mcp/tests/observability/test_wiring.py @@ -25,7 +25,9 @@ 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: + 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() diff --git a/packages/mcp/tests/test_server.py b/packages/mcp/tests/test_server.py index 8a600260..4ffea5fc 100644 --- a/packages/mcp/tests/test_server.py +++ b/packages/mcp/tests/test_server.py @@ -110,7 +110,7 @@ def test_build_pipefy_mcp_server_passes_log_level_to_fastmcp(mocked_runtime): @pytest.mark.unit def test_run_server_stdio_does_not_configure_structured_logging(monkeypatch): - """Stdio must not carry a stdout log handler: stdout is the JSON-RPC wire.""" + """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 ( From 52bb9a8e9b6364c59db9ed681f24eff1af1da8b6 Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Date: Thu, 9 Jul 2026 14:34:46 -0300 Subject: [PATCH 09/11] feat(mcp): honor inbound x-request-id and x-correlation-id Prefer an upstream request or correlation id when present so hosted request logs stay correlated across service boundaries; mint a UUID only when both headers are absent or blank. --- packages/mcp/AGENTS.md | 3 + .../observability/request_log_middleware.py | 20 +++++- .../test_request_log_middleware.py | 68 ++++++++++++++++++- 3 files changed, 89 insertions(+), 2 deletions(-) diff --git a/packages/mcp/AGENTS.md b/packages/mcp/AGENTS.md index 893258a2..f2c233eb 100644 --- a/packages/mcp/AGENTS.md +++ b/packages/mcp/AGENTS.md @@ -96,6 +96,9 @@ 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 registration diff --git a/packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py b/packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py index dde65bca..6bec0514 100644 --- a/packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py +++ b/packages/mcp/src/pipefy_mcp/observability/request_log_middleware.py @@ -27,6 +27,7 @@ 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: @@ -45,6 +46,23 @@ def _header_from_response(headers: list[tuple[bytes, bytes]], name: str) -> str 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: @@ -80,7 +98,7 @@ async def __call__( return started_at = time.perf_counter() - request_id = str(uuid4()) + request_id = resolve_request_id(scope) scope.setdefault("state", {})["request_id"] = request_id status: int | None = None diff --git a/packages/mcp/tests/observability/test_request_log_middleware.py b/packages/mcp/tests/observability/test_request_log_middleware.py index cadb8eaa..495d4f87 100644 --- a/packages/mcp/tests/observability/test_request_log_middleware.py +++ b/packages/mcp/tests/observability/test_request_log_middleware.py @@ -17,7 +17,10 @@ configure_observability_logging, reset_observability_logging, ) -from pipefy_mcp.observability.request_log_middleware import RequestLogMiddleware +from pipefy_mcp.observability.request_log_middleware import ( + RequestLogMiddleware, + resolve_request_id, +) ASGIApp = Callable[ [ @@ -247,6 +250,69 @@ async def app(scope, receive, send): 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() From e707d9e73e6282bf84a5352a4acd5a288cd2c90e Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Date: Thu, 9 Jul 2026 15:43:07 -0300 Subject: [PATCH 10/11] fix(mcp): pin structured logger at INFO independent of text level PIPEFY_MCP_LOG_LEVEL now governs only the FastMCP root logger. The hosted structured emitter stays at INFO so quieting noisy text logs cannot silently drop request/tool debugging lines. Docs frame these events as debugging with a privacy allowlist, not an audit trail. --- docs/config.md | 2 +- packages/mcp/AGENTS.md | 21 +++++--- .../pipefy_mcp/observability/json_logging.py | 24 ++++++--- packages/mcp/src/pipefy_mcp/server.py | 2 +- packages/mcp/src/pipefy_mcp/settings.py | 5 +- .../tests/observability/test_json_logging.py | 51 ++++++++++++------- .../test_request_log_middleware.py | 2 +- packages/mcp/tests/test_server.py | 2 +- 8 files changed, 71 insertions(+), 38 deletions(-) diff --git a/docs/config.md b/docs/config.md index aaa82473..a3d58685 100644 --- a/docs/config.md +++ b/docs/config.md @@ -128,4 +128,4 @@ These variables load into `pipefy_mcp.McpSettings` (`settings.mcp`). TOML keys u | `PIPEFY_MCP_HOST` | `127.0.0.1` | HTTP bind host (loopback only until hosted OBO lands). | | `PIPEFY_MCP_PORT` | `8000` | HTTP bind port. | | `PIPEFY_MCP_UNIFIED_ENVELOPE` | `true` | When true, migrated tools return `{success, data, ...}`. | -| `PIPEFY_MCP_LOG_LEVEL` | `INFO` | Governs **both** hosted structured JSON lines on **stderr** 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. Structured lines use stderr so they never share the stdio JSON-RPC stdout channel. | +| `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 f2c233eb..d5387927 100644 --- a/packages/mcp/AGENTS.md +++ b/packages/mcp/AGENTS.md @@ -82,15 +82,21 @@ shares the client's disk (remote-safe file inputs are separate follow-up work). ## Hosted structured logging -The HTTP transport emits one JSON line per request on stderr for hosted -observability (`pipefy_mcp/observability/`). Stdio does **not** install request -logging: under stdio, stdout is the JSON-RPC wire, and the structured emitter is -HTTP-only so local installs never arm that process-global handler. +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, @@ -98,7 +104,8 @@ 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. +across service boundaries. Tool lines go through the same emitter builders as +HTTP lines (`build_tool_call_event` / `emit_structured_event`). ## Tool registration @@ -242,8 +249,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 fd7bf653..0e00c3d0 100644 --- a/packages/mcp/src/pipefy_mcp/observability/json_logging.py +++ b/packages/mcp/src/pipefy_mcp/observability/json_logging.py @@ -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 stderr 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.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 @@ -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/server.py b/packages/mcp/src/pipefy_mcp/server.py index 6be2a9d9..b4157962 100644 --- a/packages/mcp/src/pipefy_mcp/server.py +++ b/packages/mcp/src/pipefy_mcp/server.py @@ -178,7 +178,7 @@ async def _serve_streamable_http(app: FastMCP, settings: Settings) -> None: """ import uvicorn - configure_observability_logging(log_level=settings.mcp.log_level) + configure_observability_logging() http_app = wire_hosted_observability(app, settings) mcp = settings.mcp config = uvicorn.Config( diff --git a/packages/mcp/src/pipefy_mcp/settings.py b/packages/mcp/src/pipefy_mcp/settings.py index dadfce9f..2b4272ac 100644 --- a/packages/mcp/src/pipefy_mcp/settings.py +++ b/packages/mcp/src/pipefy_mcp/settings.py @@ -113,8 +113,9 @@ def settings_customise_sources( log_level: McpLogLevel = Field( default="INFO", description=( - "Log level for hosted structured JSON events on stderr 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/observability/test_json_logging.py b/packages/mcp/tests/observability/test_json_logging.py index 5494edfd..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: @@ -148,7 +152,7 @@ def _isolated_logger(self): reset_observability_logging() def test_emits_valid_json_one_liner_on_stderr(self, capsys): - configure_observability_logging(log_level="INFO") + configure_observability_logging() event = build_http_request_event( method="POST", path="/mcp", @@ -170,22 +174,29 @@ def test_emits_valid_json_one_liner_on_stderr(self, capsys): 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) captured = capsys.readouterr() assert captured.out == "" - assert captured.err == "" + assert json.loads(captured.err.strip())["request_id"] == "req-pinned" def test_does_not_propagate_to_root_logger(self): root = logging.getLogger() @@ -199,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", @@ -219,15 +230,17 @@ def emit(self, record: logging.LogRecord) -> None: root.removeHandler(root_handler) def test_handler_targets_stderr_explicitly(self): - configure_observability_logging(log_level="INFO") + configure_observability_logging() logger = logging.getLogger(OBSERVABILITY_LOGGER_NAME) assert len(logger.handlers) == 1 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( diff --git a/packages/mcp/tests/observability/test_request_log_middleware.py b/packages/mcp/tests/observability/test_request_log_middleware.py index 495d4f87..18004840 100644 --- a/packages/mcp/tests/observability/test_request_log_middleware.py +++ b/packages/mcp/tests/observability/test_request_log_middleware.py @@ -88,7 +88,7 @@ def _isolated_observability_logger(): def _configure_for_capture() -> None: - configure_observability_logging(log_level="INFO") + configure_observability_logging() @pytest.mark.anyio diff --git a/packages/mcp/tests/test_server.py b/packages/mcp/tests/test_server.py index 4ffea5fc..348e5b01 100644 --- a/packages/mcp/tests/test_server.py +++ b/packages/mcp/tests/test_server.py @@ -403,7 +403,7 @@ async def test_serve_streamable_http_disables_uvicorn_access_log(remote_rs_env): mock_server_cls.return_value.serve = AsyncMock() await _serve_streamable_http(fake_app, settings) - mock_configure.assert_called_once_with(log_level="INFO") + 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, From 323206ebacda13b646ad8e6e84b4bf698c4aa340 Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Date: Thu, 9 Jul 2026 15:43:07 -0300 Subject: [PATCH 11/11] refactor(mcp): route tool_log_middleware through the structured emitter Reuse build_tool_call_event/emit_structured_event so #378 tool lines share the pinned stderr JSON channel and allowlisted shape (including cancelled and elicitation outcomes) instead of a parallel logger.info path. --- .../observability/tool_log_middleware.py | 67 ++++------- .../observability/test_tool_log_middleware.py | 113 ++++++++++-------- 2 files changed, 86 insertions(+), 94 deletions(-) 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/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 == ""