Wrapping

wrapping

Tool-call wrapping, tracing, and per-key locking runtime.

Builds the synchronous and async wrapper closures handed to FastMCP for each discovered tool: kwarg unwrapping, implicit-path warnings, external session tracing, ToolResult flattening, and per-key concurrency locking (active only in HTTP mode).

This module is a leaf — it imports only axm_mcp.concurrency at runtime. Shared structural protocols (ToolEntry, ToolLike, PlainTool, ToolResultLike) live in axm_mcp.discovery and are referenced here under TYPE_CHECKING (annotations are strings via from __future__ import annotations) plus string-literal cast targets, keeping the runtime import edge one-directional (discovery -> wrapping).

build_wrappers(name, tool)

Build the (sync, async) wrapper pair for one tool.

The single construction seam shared by the direct MCP registration path (:func:axm_mcp.discovery.register_one) and the facade path (:class:axm_mcp.facade.catalog.ToolCatalog). Both invoke the same wrappers, so kwarg-unwrapping, implicit-path warnings, tracing, exception flattening and per-key locking are invariant regardless of whether a tool is reached directly or via axm_call — there is one execution path.

Returns:

Type	Description
_SyncWrapper	(sync_wrapper, async_wrapper) where the sync wrapper carries the
_AnyWrapper	trace/flatten/exception contract and the async wrapper adds the HTTP
tuple[_SyncWrapper, _AnyWrapper]	to_thread offload plus the optional per-key lock.

Source code in packages/axm-mcp/src/axm_mcp/wrapping.py

def build_wrappers(name: str, tool: ToolEntry) -> tuple[_SyncWrapper, _AnyWrapper]:
    """Build the ``(sync, async)`` wrapper pair for one tool.

    The single construction seam shared by the direct MCP registration path
    (:func:`axm_mcp.discovery.register_one`) and the facade path
    (:class:`axm_mcp.facade.catalog.ToolCatalog`). Both invoke the *same*
    wrappers, so kwarg-unwrapping, implicit-path warnings, tracing, exception
    flattening and per-key locking are invariant regardless of whether a tool
    is reached directly or via ``axm_call`` — there is one execution path.

    Returns:
        ``(sync_wrapper, async_wrapper)`` where the sync wrapper carries the
        trace/flatten/exception contract and the async wrapper adds the HTTP
        ``to_thread`` offload plus the optional per-key lock.
    """
    is_plain = callable(tool) and not hasattr(tool, "execute")
    # Protocol tools already trace via orchestrator.run_tool()
    ctx = _WrapperCtx(name=name, should_trace=not name.startswith("protocol_"))
    sync_wrapper = (
        _build_plain_wrapper(ctx, tool) if is_plain else _build_tool_wrapper(ctx, tool)
    )
    exec_doc = getattr(getattr(tool, "execute", tool), "__doc__", None)
    sync_wrapper.__doc__ = exec_doc or f"Execute {name} tool."
    async_wrapper = _wrap_with_lock(sync_wrapper, name)
    return sync_wrapper, async_wrapper

flatten_result(result)

Flatten a ToolResult into a JSON-friendly dict.

Spreads result.data first, then sets the envelope keys (success/error/hint) deterministically. Any reserved key already present in result.data is relocated to data_{key} (with a warning) so the envelope is never clobbered and the data value is never silently lost.

Source code in packages/axm-mcp/src/axm_mcp/wrapping.py

def flatten_result(result: ToolResultLike) -> dict[str, object]:
    """Flatten a ToolResult into a JSON-friendly dict.

    Spreads ``result.data`` first, then sets the envelope keys
    (``success``/``error``/``hint``) deterministically. Any reserved key
    already present in ``result.data`` is relocated to ``data_{key}`` (with a
    warning) so the envelope is never clobbered and the data value is never
    silently lost.
    """
    output: dict[str, object] = dict(getattr(result, "data", None) or {})
    for key in _RESERVED_KEYS:
        if key in output:
            namespaced = f"data_{key}"
            logger.warning(
                "ToolResult.data key %r collides with the envelope; relocating to %r",
                key,
                namespaced,
            )
            output[namespaced] = output.pop(key)
    # ``success`` missing → False (never silently promoted to a passing result).
    output["success"] = bool(getattr(result, "success", False))
    if getattr(result, "error", None):
        output["error"] = result.error
    hint = getattr(result, "hint", None)
    if hint:
        output["hint"] = hint
    return output

log_external_step(tool_name, tool_args, success, result_str, duration_ms)

Instrumentation seam for non-protocol tool calls.

Currently a no-op. This is the hook point where an execution engine can observe each MCP tool call (name, args, outcome, duration). The legacy axm-engine tracing wiring was removed when engine was deprecated; a future axm-loom-based tracer should re-attach here. Any implementation MUST swallow its own errors — tracing must never break tool execution.

Source code in packages/axm-mcp/src/axm_mcp/wrapping.py

def log_external_step(
    tool_name: str,
    tool_args: dict[str, object],
    success: bool,
    result_str: str,
    duration_ms: int,
) -> None:
    """Instrumentation seam for non-protocol tool calls.

    Currently a no-op. This is the hook point where an execution engine
    can observe each MCP tool call (name, args, outcome, duration). The
    legacy ``axm-engine`` tracing wiring was removed when engine was
    deprecated; a future ``axm-loom``-based tracer should re-attach here.
    Any implementation MUST swallow its own errors — tracing must never
    break tool execution.
    """