From 8285edb6de72fe9ed1c40552b197a1fd87b5d6b6 Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Date: Tue, 7 Jul 2026 19:46:52 -0300 Subject: [PATCH 1/4] 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 2/4] 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 3/4] 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 4/4] 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(