diff --git a/packages/mcp/AGENTS.md b/packages/mcp/AGENTS.md
index 70f169d0..b5a590d3 100644
--- a/packages/mcp/AGENTS.md
+++ b/packages/mcp/AGENTS.md
@@ -169,3 +169,65 @@ than a local `file_path`), enforce that in the tool body gated on
 `settings.mcp.profile == "remote"` at call time, not via the marker. A tool whose
 exclusion deserves a reason gets a plain code comment stating why; the exclusion
 itself needs no annotation.
+
+## Tool-call middleware
+
+Cross-cutting concerns that wrap a tool invocation (logging, per-user quotas,
+rate limiting, cost weighting, downstream 429/circuit-breaking) register as
+ordered middleware, not by overwriting the server's internal handler. The MCP SDK
+dispatches every tool call through one `request_handlers[CallToolRequest]` slot;
+`core/tool_middleware.py` wraps that slot once, at build time, and composes the
+registered middleware around it. The middleware chain is the extension surface;
+the private slot is wrapped, not written to directly.
+
+A middleware is a plain async callable. A built-in middleware joins the per-profile
+defaults (`default_tool_middlewares` in `server.py`); a consumer of
+`build_pipefy_mcp_server` passes its own through `extra_tool_middlewares`, which the
+builder folds into the single install after the built-ins (so the default
+observability layer stays outermost). Neither path touches FastMCP internals:
+
+```python
+from pipefy_mcp.core.tool_middleware import ToolCallContext, CallNext, short_circuit_error
+
+async def quota(ctx: ToolCallContext, call_next: CallNext):
+    if over_quota(ctx.identity.client_id):
+        return short_circuit_error("quota exceeded", code="RATE_LIMITED")
+    return await call_next(ctx)
+
+# a serving layer registers its own middleware through the public builder:
+#   app = build_pipefy_mcp_server(settings, extra_tool_middlewares=[quota])
+```
+
+- **Order**: list order runs outer to inner around the tool. `[A, B]` runs A,
+  then B, then the tool, and unwinds in reverse.
+- **Short-circuit**: a middleware that returns without awaiting `call_next` skips
+  the inner chain and the tool. Use `short_circuit_error`, which carries the
+  canonical `tool_error` envelope but sets `isError=True` deliberately: a
+  governance stop means the tool never ran, distinct from a tool that ran and
+  reported a business error (`isError=False`).
+- **Identity** (`ctx.identity`): the validated caller's `client_id` and `scopes`,
+  read off the request's bearer, never re-decoded. Empty under stdio/local (no
+  inbound bearer). The end-user `subject` is intentionally absent until its
+  consumer (per-user quotas) exists.
+- **`request_id`**: correlates a call to its HTTP request when available, else the
+  JSON-RPC message id, which is client-chosen and only unique within a session.
+- **Raw arguments**: FastMCP registers the terminal with `validate_input=False`
+  and coerces arguments downstream, so middleware sees the un-coerced, client-sent
+  arguments. `ctx.argument_keys` is bounded (count and length caps) and values-free
+  for privacy-sensitive consumers; `ctx.arguments` values are passed unbounded to
+  any consumer that opts to read them. Never log a bearer or argument values.
+
+The chain installs on every profile (a no-op when the list is empty); the
+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.
+
+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
+a separate seam from the argument-validation envelope
+(`tools/validation_envelope.py`), which patches `Tool.run` to reshape a pydantic
+`ValidationError` inside FastMCP's executor: that error's structured detail exists
+only there, below this chain, so the two are complementary, not interchangeable.
diff --git a/packages/mcp/src/pipefy_mcp/auth/request_identity.py b/packages/mcp/src/pipefy_mcp/auth/request_identity.py
index 06008435..5570ac58 100644
--- a/packages/mcp/src/pipefy_mcp/auth/request_identity.py
+++ b/packages/mcp/src/pipefy_mcp/auth/request_identity.py
@@ -18,15 +18,65 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass, field
+
 from mcp.server.auth.middleware.auth_context import AuthenticatedUser
 from starlette.requests import Request
 
 
+@dataclass(frozen=True)
+class CallerIdentity:
+    """The validated caller a tool-call middleware acts on behalf of.
+
+    Sourced from the request's validated bearer, never re-decoded. ``client_id``
+    is the OAuth client (``azp``/``client_id``, falling back to ``sub`` for a
+    user token) and ``scopes`` its granted scopes. Both default empty: under the
+    stdio/local profile there is no inbound bearer, so middleware sees an
+    anonymous identity rather than a failure.
+
+    The end-user subject is intentionally deferred: its consumer (per-user
+    quotas) is not built yet. Both profiles can source it when it lands, the
+    remote profile from the validated ``sub`` claim and the local profile from
+    the configured JWT credential.
+    """
+
+    client_id: str | None = None
+    scopes: tuple[str, ...] = field(default_factory=tuple)
+
+
+def _authenticated_user(request: Request | None) -> AuthenticatedUser | None:
+    """Return the RS-validated user off the request, or None.
+
+    Reads ``request.scope`` directly rather than ``request.user``: the latter
+    asserts ``AuthenticationMiddleware`` is installed, which holds only under the
+    ``remote`` profile, so touching the property would raise on a ``local`` HTTP
+    request. This is the one place both request-scoped readers agree on how the
+    validated caller is located: off the per-message request, never
+    ``auth_context_var``.
+    """
+    user = request.scope.get("user") if request is not None else None
+    return user if isinstance(user, AuthenticatedUser) else None
+
+
+def caller_identity(request: Request | None) -> CallerIdentity:
+    """Return the validated caller off the in-flight request, or an anonymous one.
+
+    Runs on every profile (a tool-call middleware wraps both transports), so a
+    missing or non-authenticated user yields the empty :class:`CallerIdentity`
+    rather than an error.
+    """
+    user = _authenticated_user(request)
+    if user is None:
+        return CallerIdentity()
+    token = user.access_token
+    return CallerIdentity(client_id=token.client_id, scopes=tuple(token.scopes))
+
+
 def require_request_bearer(request: Request | None) -> str:
     """Return the validated bearer token off the in-flight request.
 
-    The resource-server middleware sets ``request.user`` to the
-    ``AuthenticatedUser`` it validated; this reads its access token. Taking the
+    The resource-server middleware sets the validated ``AuthenticatedUser`` on
+    the request; this reads its access token. Taking the
     request as an argument (the runtime passes ``ctx.request_context.request`` from
     the tool handler) keeps the read on the current caller: it never touches
     ``auth_context_var``, which stateful Streamable HTTP freezes at the session's
@@ -34,15 +84,14 @@ def require_request_bearer(request: Request | None) -> str:
 
     Raises when no validated bearer is present (called outside the resource-server
     request scope, or the request bore no ``AuthenticatedUser``), so a missing
-    identity fails loudly instead of issuing an unauthenticated Pipefy call. The
-    ``isinstance`` guard keeps that fail-loud property on a typed contract: an
-    unauthenticated request carries a different user type, not an
-    ``AuthenticatedUser`` with an empty token.
+    identity fails loudly instead of issuing an unauthenticated Pipefy call.
     """
-    user = request.user if request is not None else None
-    if not isinstance(user, AuthenticatedUser) or not user.access_token.token:
+    user = _authenticated_user(request)
+    if user is None or not user.access_token.token:
         raise RuntimeError(
-            "No authenticated access token on the request; the resource-server "
-            "profile must validate a bearer before a tool runs."
+            "No authenticated access token on the request. Under the "
+            "resource-server profile this means either the caller sent no valid "
+            "bearer or the authentication middleware is not wired into the ASGI "
+            "stack; a tool must not run without a validated caller."
         )
     return user.access_token.token
diff --git a/packages/mcp/src/pipefy_mcp/core/tool_middleware.py b/packages/mcp/src/pipefy_mcp/core/tool_middleware.py
new file mode 100644
index 00000000..3ed8c1dc
--- /dev/null
+++ b/packages/mcp/src/pipefy_mcp/core/tool_middleware.py
@@ -0,0 +1,259 @@
+"""Ordered middleware chain around MCP tool invocation.
+
+The server needs a seam to run cross-cutting logic around each tool call. Some of
+it spans every deployment (observability, and downstream protection such as
+honoring the API's 429s or circuit-breaking); some is specific to the multi-tenant
+hosted profile (per-user quotas, rate limiting, and cost attribution keyed on the
+validated caller). The MCP SDK offers no such seam: FastMCP dispatches every tool
+call through a single ``CallToolRequest`` entry in the low-level server's
+``request_handlers`` dict, and the only way to observe or govern a call is to
+overwrite that one private slot, so the next feature that needs it clobbers the
+previous.
+
+This module turns that single slot into an ordered chain. The composition root
+builds the middleware list per profile and passes it to
+:func:`install_tool_call_middleware`, which wraps FastMCP's handler exactly once,
+at build time, composing the middleware around it. Middleware run outer-to-inner
+in list order, may short-circuit (return an error result without invoking the
+tool), and read the validated caller from the per-message request context.
+
+The wrap targets ``app._mcp_server.request_handlers[CallToolRequest]`` and is
+tested against ``mcp==1.25.0``. If that pin moves, re-verify the handler is still
+a single ``async def handler(req: CallToolRequest) -> ServerResult`` populated by
+``FastMCP._setup_handlers``.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Awaitable, Callable, Sequence
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+from mcp import types
+from mcp.server.lowlevel.server import request_ctx
+
+from pipefy_mcp.auth.request_identity import CallerIdentity, caller_identity
+
+if TYPE_CHECKING:
+    from mcp.server.fastmcp import FastMCP
+
+__all__ = [
+    "CallNext",
+    "ToolCallContext",
+    "ToolCallMiddleware",
+    "compose",
+    "context_from_request",
+    "install_tool_call_middleware",
+    "short_circuit_error",
+]
+
+# The bounded, values-free view of a call's arguments. Keys come straight off the
+# JSON-RPC body before any validation, so an authenticated caller controls their
+# count and length; the caps keep one call from flooding a downstream log line.
+_MAX_ARG_KEYS = 64
+_MAX_ARG_KEY_LENGTH = 200
+
+# Marks the wrapped handler so a second install on the same app fails loud instead
+# of silently stacking or dropping middleware. The marker lives on the wrapped
+# function (per app), not on a module or class global: each FastMCP instance has
+# its own handler, so a global would wrap the first app and silently skip every
+# later one built in the same process.
+_INSTALLED_MARKER = "__pipefy_tool_call_chain__"
+
+
+@dataclass(frozen=True)
+class ToolCallContext:
+    """What a tool-call middleware sees for one invocation.
+
+    ``tool_name`` and ``arguments`` are read-only views over ``req.params``, so the
+    context cannot drift from the request it wraps. ``arguments`` is the raw
+    JSON-RPC argument map: FastMCP registers the terminal with
+    ``validate_input=False`` and does its own coercion/defaulting downstream, so
+    middleware observes the un-coerced, client-sent arguments. ``argument_keys`` is
+    the bounded, values-free view a privacy-sensitive consumer (logging) should
+    prefer; ``arguments`` values are passed unbounded to any consumer that opts to
+    read them.
+
+    ``request_id`` correlates a call to its HTTP request when one is available;
+    otherwise it falls back to the JSON-RPC message id, which is client-chosen and
+    only unique within a session (see :func:`context_from_request`). ``req`` is the
+    untouched request the terminal handler is called with.
+    """
+
+    argument_keys: tuple[str, ...]
+    identity: CallerIdentity
+    request_id: str | None
+    req: types.CallToolRequest
+
+    @property
+    def tool_name(self) -> str:
+        return self.req.params.name
+
+    @property
+    def arguments(self) -> dict[str, Any] | None:
+        return self.req.params.arguments
+
+
+# A middleware calls ``call_next(ctx)`` to run the inner chain (ending at the
+# tool), or returns its own ServerResult to short-circuit without running it.
+CallNext = Callable[[ToolCallContext], Awaitable[types.ServerResult]]
+ToolCallMiddleware = Callable[
+    [ToolCallContext, CallNext], Awaitable[types.ServerResult]
+]
+
+# FastMCP's registered ``CallToolRequest`` handler: the innermost callable the
+# chain wraps.
+_TerminalHandler = Callable[[types.CallToolRequest], Awaitable[types.ServerResult]]
+
+
+def _argument_keys(arguments: dict[str, Any] | None) -> tuple[str, ...]:
+    """Sorted, values-free argument key names, bounded so one call can't flood a log."""
+    if not arguments:
+        return ()
+    keys = sorted(str(key)[:_MAX_ARG_KEY_LENGTH] for key in arguments)
+    if len(keys) > _MAX_ARG_KEYS:
+        dropped = len(keys) - _MAX_ARG_KEYS
+        keys = keys[:_MAX_ARG_KEYS]
+        keys.append(f"...+{dropped} more")
+    return tuple(keys)
+
+
+def _request_scope() -> tuple[Any | None, str | None]:
+    """Return ``(request, request_id)`` from the in-flight message context.
+
+    The low-level server sets ``request_ctx`` per JSON-RPC message before calling
+    the handler, so both are available at chain entry. ``request`` is the Starlette
+    request under HTTP (``None`` over stdio). ``request_id`` prefers an HTTP-layer
+    correlation id stamped into ``scope["state"]`` (what a request-logging ASGI
+    middleware would set) and falls back to the JSON-RPC message id.
+    """
+    try:
+        ctx = request_ctx.get()
+    except LookupError:
+        return None, None
+
+    request = ctx.request
+    request_id: Any = None
+    if request is not None:
+        # scope["state"] is a dict under Starlette, but guard the type: the default
+        # only covers an absent key, so a present non-dict would crash the read.
+        state = request.scope.get("state")
+        if isinstance(state, dict):
+            request_id = state.get("request_id")
+    if request_id is None:
+        request_id = ctx.request_id
+    return request, (str(request_id) if request_id is not None else None)
+
+
+def context_from_request(req: types.CallToolRequest) -> ToolCallContext:
+    """Build the :class:`ToolCallContext` for the in-flight call."""
+    request, request_id = _request_scope()
+    return ToolCallContext(
+        argument_keys=_argument_keys(req.params.arguments),
+        identity=caller_identity(request),
+        request_id=request_id,
+        req=req,
+    )
+
+
+def compose(
+    middlewares: Sequence[ToolCallMiddleware], terminal: _TerminalHandler
+) -> CallNext:
+    """Fold ``middlewares`` around ``terminal`` into one ``call_next``.
+
+    Registration order is outer-to-inner: ``[A, B]`` runs ``A`` then ``B`` then the
+    tool, and unwinds in reverse. A middleware that returns without awaiting
+    ``call_next`` short-circuits the rest of the chain and the tool.
+    """
+
+    async def call_terminal(ctx: ToolCallContext) -> types.ServerResult:
+        return await terminal(ctx.req)
+
+    call_next: CallNext = call_terminal
+    for middleware in reversed(middlewares):
+        call_next = _bind(middleware, call_next)
+    return call_next
+
+
+def _bind(middleware: ToolCallMiddleware, call_next: CallNext) -> CallNext:
+    """Bind one middleware ahead of ``call_next`` (own scope, no late binding)."""
+
+    async def run(ctx: ToolCallContext) -> types.ServerResult:
+        return await middleware(ctx, call_next)
+
+    return run
+
+
+def short_circuit_error(
+    message: str,
+    *,
+    code: str | None = None,
+    details: dict[str, Any] | None = None,
+) -> types.ServerResult:
+    """A short-circuit result a middleware returns instead of running the tool.
+
+    Carries the canonical ``tool_error`` envelope so a client reads a governance
+    stop (quota, rate limit) the same way it reads a tool's own failure, but sets
+    ``isError=True`` deliberately: the tool never ran, distinct from a tool that ran
+    and reported a business error (``isError=False``). Bypassing FastMCP's terminal,
+    it builds the result directly, putting the envelope in the serialized ``content``
+    (where an in-tool error carries it too) and in ``structuredContent``. The match to
+    an in-tool error is on the decoded envelope, not the byte serialization.
+    """
+    # Imported lazily: pipefy_mcp.tools.tool_error_envelope pulls the tools
+    # package, which imports the runtime, which imports this module. A top-level
+    # import would form a cycle at load time.
+    from pipefy_mcp.tools.tool_error_envelope import tool_error
+
+    envelope = tool_error(message, code=code, details=details)
+    return types.ServerResult(
+        types.CallToolResult(
+            content=[
+                types.TextContent(type="text", text=json.dumps(envelope, indent=2))
+            ],
+            structuredContent=envelope,
+            isError=True,
+        )
+    )
+
+
+def install_tool_call_middleware(
+    app: FastMCP, middlewares: Sequence[ToolCallMiddleware]
+) -> None:
+    """Wrap ``app``'s ``CallToolRequest`` handler with the composed chain, once.
+
+    A no-op when ``middlewares`` is empty: FastMCP's handler is left untouched, so
+    the default (no middleware) path pays nothing per call rather than routing every
+    tool call through a pass-through wrapper that builds and discards a context.
+
+    Installed once per app: a second install on the same app raises (build the full
+    middleware list and install it once), while a different app built later in the
+    same process wraps its own fresh handler. Raises too if FastMCP has not
+    registered the handler, so an SDK-internal change fails loud rather than
+    silently skipping the chain.
+    """
+    if not middlewares:
+        return
+
+    handlers = app._mcp_server.request_handlers
+    if types.CallToolRequest not in handlers:
+        raise RuntimeError(
+            "CallToolRequest handler missing from request_handlers; the FastMCP "
+            f"internal contract changed. Present keys: {list(handlers.keys())!r}"
+        )
+
+    terminal = handlers[types.CallToolRequest]
+    if getattr(terminal, _INSTALLED_MARKER, False):
+        raise RuntimeError(
+            "a tool-call chain is already installed on this app; build the full "
+            "middleware list and install it once"
+        )
+
+    chain = compose(middlewares, terminal)
+
+    async def chained(req: types.CallToolRequest) -> types.ServerResult:
+        return await chain(context_from_request(req))
+
+    setattr(chained, _INSTALLED_MARKER, True)
+    handlers[types.CallToolRequest] = chained
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..1d9106fd
--- /dev/null
+++ b/packages/mcp/src/pipefy_mcp/observability/__init__.py
@@ -0,0 +1,5 @@
+"""Hosted-profile observability: middleware for the tool-call chain."""
+
+from pipefy_mcp.observability.tool_log_middleware import tool_log_middleware
+
+__all__ = ["tool_log_middleware"]
diff --git a/packages/mcp/src/pipefy_mcp/observability/tool_log_middleware.py b/packages/mcp/src/pipefy_mcp/observability/tool_log_middleware.py
new file mode 100644
index 00000000..766167d0
--- /dev/null
+++ b/packages/mcp/src/pipefy_mcp/observability/tool_log_middleware.py
@@ -0,0 +1,107 @@
+"""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
+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.
+"""
+
+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"]
+
+
+def _outcome(result: types.ServerResult) -> ToolCallOutcome:
+    """Map a terminal result to an outcome.
+
+    Reads ``isError`` off the result's root defensively: an experimental
+    ``CreateTaskResult`` root has no ``isError`` and reads as ``ok`` rather than
+    raising.
+    """
+    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:
+    """Emit one structured log line around a tool call, then propagate the result.
+
+    A tool body error surfaces as a result with ``isError=True`` (FastMCP's
+    terminal turns tool exceptions into an error result), so the common path logs
+    from the returned result. The two exceptions that DO propagate through the
+    chain each get their own non-error outcome and are re-raised: ``CancelledError``
+    (client disconnect / ``notifications/cancelled``) and
+    ``UrlElicitationRequiredError`` (the client must visit a URL to continue).
+    Swallowing either would break cancellation and elicitation. ``BaseException``
+    (not ``Exception``) is caught so a ``CancelledError`` cannot skip the handler
+    and let ``finally`` emit a stale ``ok``.
+    """
+    started_at = time.perf_counter()
+    outcome: ToolCallOutcome = "ok"
+    try:
+        result = await call_next(ctx)
+        outcome = _outcome(result)
+        return result
+    except asyncio.CancelledError:
+        outcome = "cancelled"
+        raise
+    except UrlElicitationRequiredError:
+        outcome = "elicitation"
+        raise
+    except BaseException:
+        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,
+                }
+            )
diff --git a/packages/mcp/src/pipefy_mcp/server.py b/packages/mcp/src/pipefy_mcp/server.py
index fe9095a7..7a1a41bd 100644
--- a/packages/mcp/src/pipefy_mcp/server.py
+++ b/packages/mcp/src/pipefy_mcp/server.py
@@ -2,12 +2,17 @@
 
 import logging
 import textwrap
-from collections.abc import AsyncIterator, Callable
+from collections.abc import AsyncIterator, Callable, Sequence
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 
 from mcp.server.fastmcp import FastMCP
 
 from pipefy_mcp.core.runtime import McpRuntime
+from pipefy_mcp.core.tool_middleware import (
+    ToolCallMiddleware,
+    install_tool_call_middleware,
+)
+from pipefy_mcp.observability.tool_log_middleware import tool_log_middleware
 from pipefy_mcp.settings import Settings
 from pipefy_mcp.tools.registry import ToolRegistry
 from pipefy_mcp.tools.validation_envelope import install_pipefy_validation_envelope
@@ -73,7 +78,26 @@ def _register_pipefy_tools(app: FastMCP, *, remote_mode: bool) -> None:
     registry.apply_remote_profile(remote_mode=remote_mode)
 
 
-def build_pipefy_mcp_server(settings: Settings) -> FastMCP:
+def default_tool_middlewares(settings: Settings) -> list[ToolCallMiddleware]:
+    """The built-in tool-call middleware to seed for the resolved profile.
+
+    Structured tool-call logging is on by default under the hosted ``remote``
+    profile, where operators rely on it to attribute activity. This is a default,
+    not a capability boundary: the chain installs on every profile, so any
+    deployment can register its own middleware, and this could later become a
+    config toggle a local deployment opts into. Deciding the seed here, at the
+    composition root, keeps the runtime free of any dependency on a concrete
+    middleware.
+    """
+    if settings.mcp.profile == "remote":
+        return [tool_log_middleware]
+    return []
+
+
+def build_pipefy_mcp_server(
+    settings: Settings,
+    extra_tool_middlewares: Sequence[ToolCallMiddleware] = (),
+) -> FastMCP:
     """Build the FastMCP app with its tools registered once, before serving.
 
     Reads everything from the resolved ``settings`` the composition root
@@ -81,6 +105,13 @@ def build_pipefy_mcp_server(settings: Settings) -> FastMCP:
     remote-safe tool surface, and ``settings.mcp.host`` / ``settings.mcp.port`` give
     the HTTP bind (they matter only for the HTTP transport; stdio ignores them).
 
+    ``extra_tool_middlewares`` is the public registration seam for a consumer of this
+    builder (a hosted serving layer that wants per-tool metrics, say): the chain
+    installs once, so a consumer folds its middleware in here rather than reaching
+    into the private ``request_handlers`` slot or re-wrapping the handler. The
+    built-in middleware runs outer to the consumer's, so the default observability
+    layer records every call including those a consumer's middleware short-circuits.
+
     The one app-scoped :class:`McpRuntime` is built via
     :meth:`McpRuntime.for_profile`, which owns both the outbound identity and (under
     ``remote``) the inbound resource-server ``(verifier, auth)`` pair. When that pair
@@ -101,6 +132,12 @@ def build_pipefy_mcp_server(settings: Settings) -> FastMCP:
         auth=auth,
     )
     _register_pipefy_tools(app, remote_mode=settings.mcp.profile == "remote")
+    # Wrap the tool-call handler with the built-in chain plus any consumer middleware.
+    # Both transports serve this app, so tool calls over stdio and HTTP alike run
+    # through the chain; the install is a no-op when the combined list is empty.
+    install_tool_call_middleware(
+        app, [*default_tool_middlewares(settings), *extra_tool_middlewares]
+    )
     return app
 
 
diff --git a/packages/mcp/tests/auth/test_request_identity.py b/packages/mcp/tests/auth/test_request_identity.py
index 744a4635..49b33fc6 100644
--- a/packages/mcp/tests/auth/test_request_identity.py
+++ b/packages/mcp/tests/auth/test_request_identity.py
@@ -12,8 +12,15 @@
 
 import pytest
 from _rs_fixtures import authenticated_user, request_with_user
+from mcp.server.auth.middleware.auth_context import AuthenticatedUser
+from mcp.server.auth.provider import AccessToken
+from starlette.requests import Request
 
-from pipefy_mcp.auth.request_identity import require_request_bearer
+from pipefy_mcp.auth.request_identity import (
+    CallerIdentity,
+    caller_identity,
+    require_request_bearer,
+)
 
 
 @pytest.mark.unit
@@ -42,3 +49,39 @@ def test_raises_on_empty_token():
     """An access token with an empty string is treated as absent."""
     with pytest.raises(RuntimeError, match="No authenticated access token"):
         require_request_bearer(request_with_user(authenticated_user("")))
+
+
+@pytest.mark.unit
+def test_caller_identity_reads_client_id_and_scopes():
+    """The validated caller's client id and scopes come off the request's token."""
+    user = AuthenticatedUser(
+        AccessToken(token="t", client_id="acting-client", scopes=["read", "write"])
+    )
+    identity = caller_identity(request_with_user(user))
+    assert identity == CallerIdentity(
+        client_id="acting-client", scopes=("read", "write")
+    )
+
+
+@pytest.mark.unit
+def test_caller_identity_is_anonymous_without_a_request():
+    """No request (stdio, or outside the request scope) yields the empty identity."""
+    assert caller_identity(None) == CallerIdentity()
+
+
+@pytest.mark.unit
+def test_caller_identity_is_anonymous_without_a_validated_user():
+    """A request whose user is not authenticated yields the empty identity."""
+    assert caller_identity(request_with_user(None)) == CallerIdentity()
+
+
+@pytest.mark.unit
+def test_caller_identity_survives_a_scope_without_a_user_key():
+    """A local-over-HTTP request has no ``user`` in scope; reading it must not raise.
+
+    ``AuthenticationMiddleware`` runs only under the remote profile, so a
+    ``local`` HTTP request never gets a ``user`` scope key. ``request.user`` would
+    assert; ``caller_identity`` reads ``scope`` directly and returns anonymous.
+    """
+    request = Request({"type": "http", "headers": []})
+    assert caller_identity(request) == CallerIdentity()
diff --git a/packages/mcp/tests/core/test_tool_middleware.py b/packages/mcp/tests/core/test_tool_middleware.py
new file mode 100644
index 00000000..df691470
--- /dev/null
+++ b/packages/mcp/tests/core/test_tool_middleware.py
@@ -0,0 +1,290 @@
+"""Unit tests for the tool-call middleware chain.
+
+Covers the seam mechanics against a real FastMCP app: composition order,
+short-circuit, exception propagation, the short-circuit envelope shape, the
+once-per-app install, and the argument-context build.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from contextlib import contextmanager
+
+import pytest
+from mcp import types
+from mcp.server.fastmcp import FastMCP
+from mcp.server.lowlevel.server import request_ctx
+from mcp.shared.context import RequestContext
+
+from pipefy_mcp.core.tool_middleware import (
+    ToolCallContext,
+    compose,
+    context_from_request,
+    install_tool_call_middleware,
+    short_circuit_error,
+)
+from pipefy_mcp.tools.tool_error_envelope import tool_error
+
+
+def _app() -> FastMCP:
+    app = FastMCP("test")
+
+    @app.tool()
+    async def echo(x: int) -> str:
+        return f"got {x}"
+
+    return app
+
+
+def _call_tool_request(**arguments: object) -> types.CallToolRequest:
+    return types.CallToolRequest(
+        method="tools/call",
+        params=types.CallToolRequestParams(name="echo", arguments=arguments),
+    )
+
+
+def _bare_context(**arguments: object) -> ToolCallContext:
+    """A minimal context for exercising the chain without a request scope."""
+    return ToolCallContext(
+        argument_keys=tuple(sorted(arguments)),
+        identity=None,  # type: ignore[arg-type]
+        request_id=None,
+        req=_call_tool_request(**arguments),
+    )
+
+
+async def _noop(ctx: ToolCallContext, call_next):
+    """A pass-through middleware, so an install actually wraps the handler."""
+    return await call_next(ctx)
+
+
+@contextmanager
+def _message_scope(request_id: str = "req-1"):
+    """Enter a request scope for one JSON-RPC message (stdio-like: no HTTP request)."""
+    token = request_ctx.set(
+        RequestContext(
+            request_id=request_id,
+            meta=None,
+            session=None,  # type: ignore[arg-type]
+            lifespan_context=None,
+            request=None,
+        )
+    )
+    try:
+        yield
+    finally:
+        request_ctx.reset(token)
+
+
+async def _invoke(app: FastMCP, req: types.CallToolRequest) -> types.ServerResult:
+    """Run the app's (wrapped) CallToolRequest handler inside a request scope."""
+    handler = app._mcp_server.request_handlers[types.CallToolRequest]
+    with _message_scope():
+        return await handler(req)
+
+
+@pytest.mark.unit
+def test_middleware_run_outer_to_inner_in_registration_order():
+    """``[A, B]`` runs A, then B, then the tool, and unwinds in reverse."""
+    trace: list[str] = []
+
+    def recorder(label: str):
+        async def mw(ctx: ToolCallContext, call_next):
+            trace.append(f"{label}:before")
+            result = await call_next(ctx)
+            trace.append(f"{label}:after")
+            return result
+
+        return mw
+
+    app = _app()
+    install_tool_call_middleware(app, [recorder("A"), recorder("B")])
+
+    result = asyncio.run(_invoke(app, _call_tool_request(x=5)))
+
+    assert result.root.isError is False
+    assert "got 5" in result.root.content[0].text
+    assert trace == ["A:before", "B:before", "B:after", "A:after"]
+
+
+@pytest.mark.unit
+def test_middleware_short_circuits_without_running_the_tool():
+    """Returning without awaiting ``call_next`` skips inner middleware and the tool."""
+    reached_inner = False
+
+    async def deny(ctx: ToolCallContext, call_next):
+        return short_circuit_error("denied", code="DENIED")
+
+    async def inner(ctx: ToolCallContext, call_next):
+        nonlocal reached_inner
+        reached_inner = True
+        return await call_next(ctx)
+
+    app = _app()
+    install_tool_call_middleware(app, [deny, inner])
+
+    result = asyncio.run(_invoke(app, _call_tool_request(x=5)))
+
+    assert result.root.isError is True
+    assert reached_inner is False
+    assert json.loads(result.root.content[0].text)["error"]["code"] == "DENIED"
+
+
+@pytest.mark.unit
+def test_chain_propagates_cancellation():
+    """A ``CancelledError`` from the terminal propagates, not swallowed by the chain."""
+
+    async def observer(ctx: ToolCallContext, call_next):
+        return await call_next(ctx)
+
+    async def boom(req: types.CallToolRequest) -> types.ServerResult:
+        raise asyncio.CancelledError
+
+    chain = compose([observer], boom)
+
+    async def run() -> None:
+        await chain(_bare_context())
+
+    with pytest.raises(asyncio.CancelledError):
+        asyncio.run(run())
+
+
+@pytest.mark.unit
+def test_chain_propagates_arbitrary_exceptions():
+    """``compose`` adds no try/except, so a terminal exception surfaces unchanged.
+
+    Covers the ``Exception`` path (``CancelledError`` above covers ``BaseException``)
+    that framework signals like ``UrlElicitationRequiredError`` travel: the chain
+    must not intercept them.
+    """
+
+    class Boom(Exception):
+        pass
+
+    async def observer(ctx: ToolCallContext, call_next):
+        return await call_next(ctx)
+
+    async def boom(req: types.CallToolRequest) -> types.ServerResult:
+        raise Boom("propagate me")
+
+    chain = compose([observer], boom)
+
+    async def run() -> None:
+        await chain(_bare_context())
+
+    with pytest.raises(Boom, match="propagate me"):
+        asyncio.run(run())
+
+
+@pytest.mark.unit
+def test_short_circuit_error_shape():
+    """The envelope is the canonical tool_error dict in both content and structured."""
+    result = short_circuit_error("quota exceeded", code="RATE_LIMITED")
+    root = result.root
+
+    envelope = json.loads(root.content[0].text)
+    assert root.isError is True
+    assert root.structuredContent == envelope
+    assert envelope == {
+        "success": False,
+        "error": {"message": "quota exceeded", "code": "RATE_LIMITED"},
+    }
+
+
+@pytest.mark.unit
+def test_short_circuit_matches_the_in_tool_error_envelope():
+    """A short-circuit decodes to the same envelope a client reads from an in-tool error.
+
+    A governance stop should carry the same ``tool_error`` payload an agent gets when a
+    tool runs and fails, so the client needs no special-casing. The two are compared on
+    the parsed envelope, not the bytes: FastMCP serializes a dict return through its own
+    encoder (non-ASCII left raw, ``structuredContent`` unset for a schema-less tool),
+    while ``short_circuit_error`` uses ``json.dumps`` and fills ``structuredContent``.
+    Comparing decoded JSON pins the contract that matters and fails loud if either
+    encoder or the envelope shape drifts, without false-alarming on cosmetics. The one
+    intended divergence: the short-circuit is ``isError=True`` (the tool never ran). The
+    non-ASCII value below crosses the encoders' escaping difference to prove the point.
+    """
+    message, code, details = "blocked", "DENIED", {"reason": "quota (café)"}
+
+    app = FastMCP("parity")
+
+    @app.tool()
+    async def failing() -> dict:
+        return tool_error(message, code=code, details=details)
+
+    request = types.CallToolRequest(
+        method="tools/call",
+        params=types.CallToolRequestParams(name="failing", arguments={}),
+    )
+    in_tool = asyncio.run(_invoke(app, request)).root
+    short_circuit = short_circuit_error(message, code=code, details=details).root
+
+    assert json.loads(in_tool.content[0].text) == json.loads(
+        short_circuit.content[0].text
+    )
+    assert in_tool.isError is False
+    assert short_circuit.isError is True
+
+
+@pytest.mark.unit
+def test_install_with_no_middleware_leaves_the_handler_untouched():
+    """An empty chain is a no-op: FastMCP's own handler stays in place."""
+    app = _app()
+    original = app._mcp_server.request_handlers[types.CallToolRequest]
+    install_tool_call_middleware(app, [])
+    assert app._mcp_server.request_handlers[types.CallToolRequest] is original
+
+
+@pytest.mark.unit
+def test_reinstalling_on_the_same_app_raises():
+    """Install is once-per-app: a second install fails loud instead of silently
+    stacking or dropping middleware. Build the full list, install once."""
+
+    async def other(ctx: ToolCallContext, call_next):
+        return await call_next(ctx)
+
+    app = _app()
+    install_tool_call_middleware(app, [_noop])
+    with pytest.raises(RuntimeError, match="already installed"):
+        install_tool_call_middleware(app, [_noop, other])
+
+
+@pytest.mark.unit
+def test_install_wraps_each_app_independently():
+    """A second app built in the same process wraps its own handler.
+
+    Guards against a module- or class-global sentinel, which would wrap the first
+    app and silently skip every later one.
+    """
+    app_one = _app()
+    app_two = _app()
+    install_tool_call_middleware(app_one, [_noop])
+    install_tool_call_middleware(app_two, [_noop])
+
+    handler_one = app_one._mcp_server.request_handlers[types.CallToolRequest]
+    handler_two = app_two._mcp_server.request_handlers[types.CallToolRequest]
+    assert handler_one is not handler_two
+
+
+@pytest.mark.unit
+def test_install_raises_when_handler_absent():
+    """A missing CallToolRequest handler fails loud (SDK contract changed)."""
+    app = _app()
+    del app._mcp_server.request_handlers[types.CallToolRequest]
+    with pytest.raises(RuntimeError, match="CallToolRequest handler missing"):
+        install_tool_call_middleware(app, [_noop])
+
+
+@pytest.mark.unit
+def test_context_exposes_bounded_argument_keys_without_values():
+    """The context carries sorted argument keys, never the values."""
+    req = _call_tool_request(zeta=1, alpha="secret-value")
+    with _message_scope(request_id="m-1"):
+        ctx = context_from_request(req)
+
+    assert ctx.tool_name == "echo"
+    assert ctx.argument_keys == ("alpha", "zeta")
+    # No HTTP request in scope -> falls back to the JSON-RPC message id.
+    assert ctx.request_id == "m-1"
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_tool_log_middleware.py b/packages/mcp/tests/observability/test_tool_log_middleware.py
new file mode 100644
index 00000000..a340d486
--- /dev/null
+++ b/packages/mcp/tests/observability/test_tool_log_middleware.py
@@ -0,0 +1,161 @@
+"""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 __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.tool_log_middleware import tool_log_middleware
+
+
+def _context(**arguments: object) -> ToolCallContext:
+    req = types.CallToolRequest(
+        method="tools/call",
+        params=types.CallToolRequestParams(name="get_card", arguments=arguments),
+    )
+    return ToolCallContext(
+        argument_keys=tuple(sorted(arguments)),
+        identity=CallerIdentity(client_id="acting-client", scopes=("read",)),
+        request_id="req-42",
+        req=req,
+    )
+
+
+def _ok_result() -> types.ServerResult:
+    return types.ServerResult(
+        types.CallToolResult(content=[types.TextContent(type="text", text="ok")])
+    )
+
+
+def _log_line(caplog) -> dict:
+    records = [
+        r for r in caplog.records if r.name == "pipefy_mcp.observability.tool_call"
+    ]
+    assert len(records) == 1
+    return json.loads(records[0].getMessage())
+
+
+@pytest.mark.unit
+def test_logs_one_line_with_the_documented_fields(caplog):
+    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))
+
+    event = _log_line(caplog)
+    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))
+
+
+@pytest.mark.unit
+def test_never_logs_argument_values_or_a_bearer(caplog):
+    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)
+        )
+
+    raw = caplog.records[-1].getMessage()
+    assert "super-secret-bearer" not in raw
+    assert "token" in json.loads(raw)["arg_keys"]  # the key is logged, not the value
+
+
+@pytest.mark.unit
+def test_error_result_logs_outcome_error(caplog):
+    async def terminal(ctx):
+        return types.ServerResult(
+            types.CallToolResult(
+                content=[types.TextContent(type="text", text="boom")], isError=True
+            )
+        )
+
+    with caplog.at_level(logging.INFO, logger="pipefy_mcp.observability.tool_call"):
+        asyncio.run(tool_log_middleware(_context(card_id="1"), terminal))
+
+    assert _log_line(caplog)["outcome"] == "error"
+
+
+@pytest.mark.unit
+def test_cancellation_logs_outcome_cancelled_and_re_raises(caplog):
+    """A client disconnect is control flow, not an error, and must propagate."""
+
+    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))
+
+    assert _log_line(caplog)["outcome"] == "cancelled"
+
+
+@pytest.mark.unit
+def test_elicitation_logs_outcome_elicitation_and_re_raises(caplog):
+    """A URL-elicitation signal is a continuation, not an error, and must propagate."""
+
+    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))
+
+    assert _log_line(caplog)["outcome"] == "elicitation"
+
+
+@pytest.mark.unit
+def test_other_propagated_exception_logs_outcome_error_and_re_raises(caplog):
+    """Any other propagating exception is logged as an error and re-raised."""
+
+    class Boom(Exception):
+        pass
+
+    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))
+
+    assert _log_line(caplog)["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."""
+
+    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)
+
+    captured = capsys.readouterr()
+    assert captured.out == ""
+    assert '"event":"tool_call"' in captured.err
diff --git a/packages/mcp/tests/test_server.py b/packages/mcp/tests/test_server.py
index f3cc6848..d15bfc45 100644
--- a/packages/mcp/tests/test_server.py
+++ b/packages/mcp/tests/test_server.py
@@ -1,3 +1,4 @@
+import json
 from datetime import timedelta
 from unittest.mock import MagicMock, patch
 
@@ -11,11 +12,14 @@
 from pipefy_auth import AuthSettings
 from pipefy_sdk import PipefySettings
 
+from pipefy_mcp.core.tool_middleware import ToolCallContext, short_circuit_error
+from pipefy_mcp.observability.tool_log_middleware import tool_log_middleware
 from pipefy_mcp.server import (
     _assert_safe_http_bind,
     _make_lifespan,
     _register_pipefy_tools,
     build_pipefy_mcp_server,
+    default_tool_middlewares,
     run_server,
 )
 from pipefy_mcp.settings import McpSettings, Settings, resolve_mcp_settings
@@ -131,6 +135,49 @@ def test_build_server_remote_mode_exposes_only_the_remote_safe_seed(mocked_runti
     assert "execute_graphql" not in exposed
 
 
+@pytest.mark.unit
+def test_default_tool_middlewares_seeds_the_logger_under_remote():
+    """The composition root seeds structured tool-call logging for the hosted profile."""
+    assert default_tool_middlewares(_REMOTE_PROFILE_SETTINGS) == [tool_log_middleware]
+
+
+@pytest.mark.unit
+def test_default_tool_middlewares_seeds_nothing_under_local():
+    """The local profile gets no default middleware; the chain stays empty."""
+    assert default_tool_middlewares(_MINIMAL_PIPEFY_SETTINGS) == []
+
+
+@pytest.mark.anyio
+async def test_extra_tool_middlewares_register_through_the_public_builder(
+    mocked_runtime,
+):
+    """A consumer's middleware installs via ``extra_tool_middlewares``, no private reach.
+
+    Drives a real MCP session against an app built through the public builder: a
+    short-circuiting middleware passed as ``extra_tool_middlewares`` returns its
+    envelope, so the consumer's middleware demonstrably runs in the installed chain
+    (the tool never executes, so no live client is needed). The short-circuit
+    envelope's shape is pinned elsewhere; this asserts only that the seam wires it in.
+    """
+
+    async def deny(ctx: ToolCallContext, call_next):
+        return short_circuit_error("blocked by consumer", code="DENIED")
+
+    app = build_pipefy_mcp_server(
+        _MINIMAL_PIPEFY_SETTINGS, extra_tool_middlewares=[deny]
+    )
+    session = create_client_session(
+        app,
+        read_timeout_seconds=timedelta(seconds=10),
+        raise_exceptions=True,
+    )
+    async with session as s:
+        result = await s.call_tool("get_organization", {})
+
+    assert result.isError is True
+    assert json.loads(result.content[0].text)["error"]["code"] == "DENIED"
+
+
 @pytest.mark.unit
 def test_second_registration_pass_is_rejected_by_collision_preflight(mocked_runtime):
     """A second registration pass on the same app is rejected by the preflight.