feat(mcp): add JSON structured logging emitter and JWT sub claim (#373 stack 1/3)

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_<NAME>` 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. |

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,14 +110,20 @@ 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(
+        # 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,
             scopes=_parse_scopes(claims.get("scope")),
             # exp is an RFC 7519 NumericDate (may be fractional); AccessToken
             # wants an int.
             expires_at=int(exp),
             resource=self._resource,
+            sub=sub_claim if isinstance(sub_claim, str) else None,
         )
 
 

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

@@ -1,5 +1,26 @@
-"""Hosted-profile observability: middleware for the tool-call chain."""
+"""Hosted observability: structured JSON logging and tool-call middleware.
 
-from pipefy_mcp.observability.tool_log_middleware import tool_log_middleware
+Only the stdlib-only emitter surface is re-exported here; ``tool_log_middleware``
+and (later) HTTP wiring stay submodule imports so importing this package never
+pulls ``starlette`` or the MCP SDK.
+"""
 
-__all__ = ["tool_log_middleware"]
+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",
+]

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

@@ -0,0 +1,152 @@
+"""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",
+    }
+)
+
+
+_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.
+
+    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 _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"
+        ) 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,
+    }

packages/mcp/src/pipefy_mcp/server.py

@@ -12,6 +12,7 @@
     ToolCallMiddleware,
     install_tool_call_middleware,
 )
+from pipefy_mcp.observability.json_logging import configure_observability_logging
 from pipefy_mcp.observability.tool_log_middleware import tool_log_middleware
 from pipefy_mcp.settings import Settings
 from pipefy_mcp.tools.registry import ToolRegistry
@@ -128,6 +129,7 @@ def build_pipefy_mcp_server(
         lifespan=_make_lifespan(runtime),
         host=settings.mcp.host,
         port=settings.mcp.port,
+        log_level=settings.mcp.log_level,
         token_verifier=verifier,
         auth=auth,
     )
@@ -193,6 +195,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)

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

packages/mcp/tests/auth/test_resource_server.py

@@ -50,11 +50,43 @@ 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_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(