Index

facade

AXM MCP facade — compact tool surface over standard MCP primitives.

Instead of registering every discovered axm.tools entry point as an individual MCP tool (one JSON-Schema each in tools/list), the facade exposes four meta-tools — axm_search / axm_describe / axm_call / axm_capabilities — plus a small hot path of frequently-used tools that opt in via expose_directly = True. Everything else is reachable through axm_search -> axm_describe -> axm_call, keeping the per-session tools/list payload small without depending on any client-specific deferred-loading mechanism.

The catalog is built from the same discover_tools() dict the server already holds — it does not re-import tools.

ToolCatalog

Searchable index over discovered axm.tools entries.

Parameters:

Name	Type	Description	Default
tools	dict[str, ToolEntry]	The discover_tools() mapping (name -> tool entry). The catalog stores it by reference; it does not re-discover.	required

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

class ToolCatalog:
    """Searchable index over discovered ``axm.tools`` entries.

    Args:
        tools: The ``discover_tools()`` mapping (name -> tool entry).  The
            catalog stores it by reference; it does not re-discover.
    """

    def __init__(self, tools: dict[str, ToolEntry]) -> None:
        self._entries = tools
        # Build the SAME wrapper pair the direct MCP path uses, so
        # ``axm_call`` runs through the identical lock/trace/flatten contract
        # (there is a single execution path). Built lazily-eager here so
        # discovery cost is paid once, at catalog construction.
        self._wrappers: dict[str, tuple[_SyncWrapper, _AnyWrapper]] = {
            name: build_wrappers(name, tool) for name, tool in tools.items()
        }

    # ── introspection ────────────────────────────────────────────────────

    def names(self) -> list[str]:
        """All tool names, sorted."""
        return sorted(self._entries)

    def hot_path(self) -> list[str]:
        """Names of tools opting into direct MCP exposure (``expose_directly``)."""
        return sorted(
            name
            for name, tool in self._entries.items()
            if tool_metadata(tool).expose_directly
        )

    # ── facade operations ────────────────────────────────────────────────

    def search(
        self,
        query: str,
        domain: str | None = None,
        limit: int = _DEFAULT_LIMIT,
    ) -> list[dict[str, object]]:
        """Find tools whose name, summary, tags or domain match *query*.

        Matching is case-insensitive substring (the deliberately-simple v1;
        tags carry discovery quality).  An empty *query* lists everything
        (optionally scoped by *domain*), which makes the facade browsable.

        Args:
            query: Substring to look for. Empty matches all.
            domain: Optional exact-match domain filter.
            limit: Maximum number of results.

        Returns:
            A list of ``{name, summary, domain, tags}`` dicts, name-sorted.
        """
        q = query.lower().strip()
        results: list[dict[str, object]] = []
        for name in sorted(self._entries):
            tool = self._entries[name]
            meta = tool_metadata(tool)
            if domain is not None and meta.domain != domain:
                continue
            summary = _summary(tool)
            haystack = f"{name}\n{summary}\n{' '.join(meta.tags)}\n{meta.domain or ''}"
            if q and q not in haystack.lower():
                continue
            results.append(
                {
                    "name": name,
                    "summary": summary,
                    "domain": meta.domain,
                    "tags": sorted(meta.tags),
                }
            )
            if len(results) >= limit:
                break
        return results

    def describe(self, name: str) -> dict[str, object]:
        """Return the full invocation contract for *name*.

        Args:
            name: Tool name.

        Returns:
            ``{name, summary, domain, tags, docstring, params}`` where each
            param is ``{name, annotation, required, default}``.

        Raises:
            UnknownToolError: If *name* is not in the catalog.
        """
        tool = self._get(name)
        meta = tool_metadata(tool)
        return {
            "name": name,
            "summary": _summary(tool),
            "domain": meta.domain,
            "tags": sorted(meta.tags),
            "docstring": _doc(tool),
            "params": [dataclasses.asdict(p) for p in self._params(name)],
        }

    def call(self, name: str, arguments: dict[str, object] | None = None) -> str:
        """Execute *name* synchronously and return its text output.

        Delegates to the *same* sync wrapper the direct MCP path registers
        (built by :func:`axm_mcp.wrapping.build_wrappers`), so ``axm_call``
        inherits its full contract: kwarg unwrapping, implicit-path warning,
        tracing, exception flattening (a raised ``ValueError``/``OSError``
        becomes ``{success: False, error: ...}`` — never a raw MCP protocol
        error), and the success-only ``text`` short-circuit. This is the
        single execution path.

        The per-key concurrency lock (``git_*``/``protocol_*`` in HTTP mode)
        is only held via :meth:`acall`; ``call`` is the sync entry used in
        stdio mode and by tests.

        Args:
            name: Tool name.
            arguments: Keyword arguments for the tool.

        Returns:
            The tool's ``ToolResult.text`` when the result is successful, else a
            readable rendering of the shared flattened envelope.

        Raises:
            UnknownToolError: If *name* is not in the catalog.
        """
        tool = self._get(name)  # validate name → UnknownToolError
        self._bind_check(tool, arguments)
        sync_wrapper, _ = self._wrappers[name]
        return self._render(sync_wrapper(**(arguments or {})))

    async def acall(self, name: str, arguments: dict[str, object] | None = None) -> str:
        """Execute *name* through the lock-aware async wrapper.

        The HTTP-mode entry point: it awaits the async wrapper, so the
        per-key lock (``git_*`` by repo path, ``protocol_*`` by session) is
        held and the sync tool body runs on a worker thread via
        ``asyncio.to_thread`` — the event loop of the shared server is never
        blocked. In stdio mode the async wrapper simply runs the tool inline.

        Args:
            name: Tool name.
            arguments: Keyword arguments for the tool.

        Returns:
            The rendered text output (same contract as :meth:`call`).

        Raises:
            UnknownToolError: If *name* is not in the catalog.
        """
        tool = self._get(name)  # validate name → UnknownToolError
        self._bind_check(tool, arguments)
        _, async_wrapper = self._wrappers[name]
        result = await async_wrapper(**(arguments or {}))
        return self._render(result)

    @staticmethod
    def _bind_check(tool: ToolEntry, arguments: dict[str, object] | None) -> None:
        """Bind *arguments* to the tool's signature, raising ``TypeError`` early.

        Separates a genuine **argument mismatch** (missing required / unexpected
        keyword — the loss of client-side schema validation that ``axm_call``
        must compensate with a param hint) from a ``TypeError`` raised *inside*
        the tool body (which the wrapper catches and flattens). Only the former
        propagates here as a ``TypeError``; a dispatcher's ``**kwargs`` makes
        the bind permissive, so those tools never false-positive.
        """
        try:
            sig = inspect.signature(_exec_fn(tool))
        except (ValueError, TypeError):
            return  # un-introspectable signature — let the wrapper handle it
        sig.bind(**(arguments or {}))

    @staticmethod
    def _render(result: dict[str, object] | str) -> str:
        """Render a wrapper result (already flattened) to facade text."""
        if isinstance(result, str):
            return result
        return "\n".join(f"{k}: {v}" for k, v in result.items())

    def param_hint(self, name: str) -> str:
        """Human-readable list of accepted params for *name* (for error text)."""
        try:
            params = self._params(name)
        except UnknownToolError:
            return ""
        parts = [
            f"{p.name}: {p.annotation}" + ("" if p.required else f" = {p.default}")
            for p in params
        ]
        return ", ".join(parts)

    def _params(self, name: str) -> list[_ParamSpec]:
        """Typed parameter specs for *name* (shared by describe/param_hint)."""
        return [
            _ParamSpec(
                name=p.name,
                annotation=_annotation_str(p.annotation),
                required=p.default is inspect.Parameter.empty,
                default=(
                    None if p.default is inspect.Parameter.empty else repr(p.default)
                ),
            )
            for p in signature_params(_exec_fn(self._get(name)))
        ]

    def capabilities(self, domain: str | None = None) -> dict[str, list[str]]:
        """Group tool names by domain.

        Args:
            domain: When given, return only that domain's tools.

        Returns:
            ``{domain: [tool names]}``; tools without a domain are grouped
            under the ``"(ungrouped)"`` key.
        """
        groups: dict[str, list[str]] = {}
        for name in sorted(self._entries):
            d = tool_metadata(self._entries[name]).domain or "(ungrouped)"
            if domain is not None and d != domain:
                continue
            groups.setdefault(d, []).append(name)
        return groups

    # ── internals ────────────────────────────────────────────────────────

    def _get(self, name: str) -> ToolEntry:
        try:
            return self._entries[name]
        except KeyError as exc:
            known = ", ".join(sorted(self._entries))
            msg = f"Unknown tool {name!r}. Known tools: {known or '<none>'}"
            raise UnknownToolError(msg) from exc

acall(name, arguments=None) async

Execute name through the lock-aware async wrapper.

The HTTP-mode entry point: it awaits the async wrapper, so the per-key lock (git_* by repo path, protocol_* by session) is held and the sync tool body runs on a worker thread via asyncio.to_thread — the event loop of the shared server is never blocked. In stdio mode the async wrapper simply runs the tool inline.

Parameters:

Name	Type	Description	Default
name	str	Tool name.	required
arguments	dict[str, object] | None	Keyword arguments for the tool.	None

Returns:

Type	Description
str	The rendered text output (same contract as :meth:call).

Raises:

Type	Description
UnknownToolError	If name is not in the catalog.

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

async def acall(self, name: str, arguments: dict[str, object] | None = None) -> str:
    """Execute *name* through the lock-aware async wrapper.

    The HTTP-mode entry point: it awaits the async wrapper, so the
    per-key lock (``git_*`` by repo path, ``protocol_*`` by session) is
    held and the sync tool body runs on a worker thread via
    ``asyncio.to_thread`` — the event loop of the shared server is never
    blocked. In stdio mode the async wrapper simply runs the tool inline.

    Args:
        name: Tool name.
        arguments: Keyword arguments for the tool.

    Returns:
        The rendered text output (same contract as :meth:`call`).

    Raises:
        UnknownToolError: If *name* is not in the catalog.
    """
    tool = self._get(name)  # validate name → UnknownToolError
    self._bind_check(tool, arguments)
    _, async_wrapper = self._wrappers[name]
    result = await async_wrapper(**(arguments or {}))
    return self._render(result)

call(name, arguments=None)

Execute name synchronously and return its text output.

Delegates to the same sync wrapper the direct MCP path registers (built by :func:axm_mcp.wrapping.build_wrappers), so axm_call inherits its full contract: kwarg unwrapping, implicit-path warning, tracing, exception flattening (a raised ValueError/OSError becomes {success: False, error: ...} — never a raw MCP protocol error), and the success-only text short-circuit. This is the single execution path.

The per-key concurrency lock (git_*/protocol_* in HTTP mode) is only held via :meth:acall; call is the sync entry used in stdio mode and by tests.

Parameters:

Name	Type	Description	Default
name	str	Tool name.	required
arguments	dict[str, object] | None	Keyword arguments for the tool.	None

Returns:

Type	Description
str	The tool's ToolResult.text when the result is successful, else a
str	readable rendering of the shared flattened envelope.

Raises:

Type	Description
UnknownToolError	If name is not in the catalog.

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

def call(self, name: str, arguments: dict[str, object] | None = None) -> str:
    """Execute *name* synchronously and return its text output.

    Delegates to the *same* sync wrapper the direct MCP path registers
    (built by :func:`axm_mcp.wrapping.build_wrappers`), so ``axm_call``
    inherits its full contract: kwarg unwrapping, implicit-path warning,
    tracing, exception flattening (a raised ``ValueError``/``OSError``
    becomes ``{success: False, error: ...}`` — never a raw MCP protocol
    error), and the success-only ``text`` short-circuit. This is the
    single execution path.

    The per-key concurrency lock (``git_*``/``protocol_*`` in HTTP mode)
    is only held via :meth:`acall`; ``call`` is the sync entry used in
    stdio mode and by tests.

    Args:
        name: Tool name.
        arguments: Keyword arguments for the tool.

    Returns:
        The tool's ``ToolResult.text`` when the result is successful, else a
        readable rendering of the shared flattened envelope.

    Raises:
        UnknownToolError: If *name* is not in the catalog.
    """
    tool = self._get(name)  # validate name → UnknownToolError
    self._bind_check(tool, arguments)
    sync_wrapper, _ = self._wrappers[name]
    return self._render(sync_wrapper(**(arguments or {})))

capabilities(domain=None)

Group tool names by domain.

Parameters:

Name	Type	Description	Default
domain	str | None	When given, return only that domain's tools.	None

Returns:

Type	Description
dict[str, list[str]]	{domain: [tool names]}; tools without a domain are grouped
dict[str, list[str]]	under the "(ungrouped)" key.

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

def capabilities(self, domain: str | None = None) -> dict[str, list[str]]:
    """Group tool names by domain.

    Args:
        domain: When given, return only that domain's tools.

    Returns:
        ``{domain: [tool names]}``; tools without a domain are grouped
        under the ``"(ungrouped)"`` key.
    """
    groups: dict[str, list[str]] = {}
    for name in sorted(self._entries):
        d = tool_metadata(self._entries[name]).domain or "(ungrouped)"
        if domain is not None and d != domain:
            continue
        groups.setdefault(d, []).append(name)
    return groups

describe(name)

Return the full invocation contract for name.

Parameters:

Name	Type	Description	Default
name	str	Tool name.	required

Returns:

Type	Description
dict[str, object]	{name, summary, domain, tags, docstring, params} where each
dict[str, object]	param is {name, annotation, required, default}.

Raises:

Type	Description
UnknownToolError	If name is not in the catalog.

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

def describe(self, name: str) -> dict[str, object]:
    """Return the full invocation contract for *name*.

    Args:
        name: Tool name.

    Returns:
        ``{name, summary, domain, tags, docstring, params}`` where each
        param is ``{name, annotation, required, default}``.

    Raises:
        UnknownToolError: If *name* is not in the catalog.
    """
    tool = self._get(name)
    meta = tool_metadata(tool)
    return {
        "name": name,
        "summary": _summary(tool),
        "domain": meta.domain,
        "tags": sorted(meta.tags),
        "docstring": _doc(tool),
        "params": [dataclasses.asdict(p) for p in self._params(name)],
    }

hot_path()

Names of tools opting into direct MCP exposure (expose_directly).

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

def hot_path(self) -> list[str]:
    """Names of tools opting into direct MCP exposure (``expose_directly``)."""
    return sorted(
        name
        for name, tool in self._entries.items()
        if tool_metadata(tool).expose_directly
    )

names()

All tool names, sorted.

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

def names(self) -> list[str]:
    """All tool names, sorted."""
    return sorted(self._entries)

param_hint(name)

Human-readable list of accepted params for name (for error text).

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

def param_hint(self, name: str) -> str:
    """Human-readable list of accepted params for *name* (for error text)."""
    try:
        params = self._params(name)
    except UnknownToolError:
        return ""
    parts = [
        f"{p.name}: {p.annotation}" + ("" if p.required else f" = {p.default}")
        for p in params
    ]
    return ", ".join(parts)

search(query, domain=None, limit=_DEFAULT_LIMIT)

Find tools whose name, summary, tags or domain match query.

Matching is case-insensitive substring (the deliberately-simple v1; tags carry discovery quality). An empty query lists everything (optionally scoped by domain), which makes the facade browsable.

Parameters:

Name	Type	Description	Default
query	str	Substring to look for. Empty matches all.	required
domain	str | None	Optional exact-match domain filter.	None
limit	int	Maximum number of results.	_DEFAULT_LIMIT

Returns:

Type	Description
list[dict[str, object]]	A list of {name, summary, domain, tags} dicts, name-sorted.

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

def search(
    self,
    query: str,
    domain: str | None = None,
    limit: int = _DEFAULT_LIMIT,
) -> list[dict[str, object]]:
    """Find tools whose name, summary, tags or domain match *query*.

    Matching is case-insensitive substring (the deliberately-simple v1;
    tags carry discovery quality).  An empty *query* lists everything
    (optionally scoped by *domain*), which makes the facade browsable.

    Args:
        query: Substring to look for. Empty matches all.
        domain: Optional exact-match domain filter.
        limit: Maximum number of results.

    Returns:
        A list of ``{name, summary, domain, tags}`` dicts, name-sorted.
    """
    q = query.lower().strip()
    results: list[dict[str, object]] = []
    for name in sorted(self._entries):
        tool = self._entries[name]
        meta = tool_metadata(tool)
        if domain is not None and meta.domain != domain:
            continue
        summary = _summary(tool)
        haystack = f"{name}\n{summary}\n{' '.join(meta.tags)}\n{meta.domain or ''}"
        if q and q not in haystack.lower():
            continue
        results.append(
            {
                "name": name,
                "summary": summary,
                "domain": meta.domain,
                "tags": sorted(meta.tags),
            }
        )
        if len(results) >= limit:
            break
    return results