MCP server

The mirror of the MCP connector. Where the connector turns an external MCP server into lazybridge.Tool entries, the MCP server does the opposite: it exposes LazyTools' own tool providers over MCP so any MCP host — Claude Desktop, Claude Code, ChatGPT Codex — can call datahub_*, statistical_*, regime_* and web search/crawl as native tools.

Status: alpha. Install: pip install "lazytoolkit[mcp] @ git+https://github.com/selvaz/LazyTools.git".

The bridge is thin because LazyBridge already normalises every tool behind one abstraction: tool.definition() yields the JSON Schema MCP wants for inputSchema, and await tool.run(**kwargs) dispatches with argument validation built in. The whole server is expand providers → list them → dispatch calls.

Quick start

Run it over stdio (the transport MCP hosts launch):

lazytools-mcp                       # all read-only providers
lazytools-mcp datahub statistical   # a subset (positional provider ids)
python -m lazytools.mcp_server      # equivalent module form

Point Claude Desktop / Claude Code at it:

{
  "mcpServers": {
    "lazytools": { "command": "lazytools-mcp" }
  }
}

That's it — the host now sees LazyTools' read-only surface as native tools. Providers whose optional extra is missing are simply skipped, so a bare [mcp] install serves datahub + statistical; install lazystats[regimes] / the [web] extra to light up regimes / web.

Providers

lazytools-mcp serves the read-only provider menu. Pass ids to select a subset, or set LAZYTOOLS_MCP_PROVIDERS=datahub,statistical in the env.

id	Provider	Tools	Needs
datahub	DataHubTools()	datahub_* discovery, resolution, financial facts	market-data-hub installed
statistical	StatisticalAnalysisTools()	volatility, correlation, outliers, regression	market-data-hub installed
regimes	RegimeTools()	regime_* read-only inspection	lazystats[regimes]
web	WebTools()	search / crawl / get-page	[web] extra

Safety model

The server is read-only by default, enforced in two layers:

1. Provider-level configuration (authoritative). default_providers() constructs every provider in its read-only shape — DataHubTools() without allow_raw_series / allow_refresh, RegimeTools() with allow_write=False, which never even emit their write tools.
2. A name-based guard (secondary). read_only=True additionally drops any tool whose name matches UNSAFE_TOOL_PATTERNS (e.g. *_send, *_write, *_delete, *_ensure_*, *_fit) with a logged warning — a coarse net for the case where a write-enabled provider is passed in by mistake.

There is no interactive ConfirmationGate over MCP, so mutating tools stay off the default surface. To expose them you must opt in explicitly:

• CLI — --allow-unsafe constructs the providers in write-enabled mode (default_providers(ids, allow_write=True)) and disables the name guard, so the menu's writers (datahub refresh/register, regime fit/persist/delete) are emitted and served. No gating is applied.
• Programmatic — build write-enabled providers yourself and pass read_only=False to build_server, ideally with your own allow-list / confirmation wrapper around the mutating tools.

Programmatic use

import asyncio
from lazytools.mcp_server import build_server, serve_stdio, default_providers

# Read-only by default.
server = build_server(default_providers())
asyncio.run(serve_stdio(server))

build_server accepts any mix of ToolProviders, Tools, and plain callables — so you can serve a custom, curated surface:

from lazytools.connectors.datahub import DataHubTools
from lazytools.statistical_analysis import StatisticalAnalysisTools

server = build_server(
    [DataHubTools(), StatisticalAnalysisTools()],
    name="lazytools-finance",
    instructions="Read-only market data + statistics.",
)

API

lazytools.mcp_server

LazyTools MCP server — expose LazyTools' tool providers over MCP.

The mirror of :mod:lazytools.connectors.mcp (the MCP client). Where the client turns an external MCP server into lazybridge.Tool entries, this server turns LazyTools' own providers into an MCP endpoint any host (Claude Desktop, Claude Code, Codex, …) can call.

Quick start (stdio, the transport MCP hosts launch)::

python -m lazytools.mcp_server            # all read-only providers
python -m lazytools.mcp_server datahub statistical   # a subset

Claude Desktop / Claude Code config::

{
  "mcpServers": {
    "lazytools": { "command": "lazytools-mcp" }
  }
}

Programmatic use::

import asyncio
from lazytools.mcp_server import build_server, serve_stdio, default_providers

server = build_server(default_providers())   # read-only by default
asyncio.run(serve_stdio(server))

Install with::

pip install lazytoolkit[mcp]

The server is read-only by default; see :func:build_server for the safety model.

PROVIDER_FACTORIES module-attribute

PROVIDER_FACTORIES: dict[str, Callable[[bool], Any]] = {}

UNSAFE_TOOL_PATTERNS module-attribute

UNSAFE_TOOL_PATTERNS: tuple[str, ...] = ('_send', '_write', '_delete', '_register', '_ensure_', '_refresh', '_persist', '_save', '_export_', '_fit')

build_server

build_server(providers: Sequence[Any], *, name: str = 'lazytools', version: str | None = None, read_only: bool = True, unsafe_patterns: Iterable[str] = UNSAFE_TOOL_PATTERNS, instructions: str | None = None) -> Server

Build a low-level MCP :class:~mcp.server.lowlevel.Server for providers.

The returned server has two handlers wired:

• list_tools → one MCP Tool per expanded LazyBridge tool, with inputSchema taken verbatim from tool.definition().parameters;
• call_tool → await tool.run(**arguments), the return value serialized via :func:result_to_text. Tool errors are returned as MCP error results (isError=True) instead of crashing the session.

Requires the mcp extra (pip install lazytoolkit[mcp]).

Source code in src/lazytools/mcp_server/server.py

def build_server(
    providers: Sequence[Any],
    *,
    name: str = "lazytools",
    version: str | None = None,
    read_only: bool = True,
    unsafe_patterns: Iterable[str] = UNSAFE_TOOL_PATTERNS,
    instructions: str | None = None,
) -> Server:
    """Build a low-level MCP :class:`~mcp.server.lowlevel.Server` for ``providers``.

    The returned server has two handlers wired:

    * ``list_tools`` → one MCP ``Tool`` per expanded LazyBridge tool, with
      ``inputSchema`` taken verbatim from ``tool.definition().parameters``;
    * ``call_tool`` → ``await tool.run(**arguments)``, the return value
      serialized via :func:`result_to_text`. Tool errors are returned as
      MCP error results (``isError=True``) instead of crashing the session.

    Requires the ``mcp`` extra (``pip install lazytoolkit[mcp]``).
    """
    try:
        import mcp.types as types
        from mcp.server.lowlevel import Server
    except ImportError as exc:  # pragma: no cover - exercised without the extra
        raise ImportError(
            "lazytools.mcp_server requires the MCP SDK: pip install 'lazytoolkit[mcp]'"
        ) from exc

    tool_map = expand_tools(providers, read_only=read_only, unsafe_patterns=unsafe_patterns)
    logger.info("LazyTools MCP server exposing %d tool(s): %s", len(tool_map), ", ".join(sorted(tool_map)))

    server: Server = Server(name, version=version, instructions=instructions)

    @server.list_tools()
    async def _list_tools() -> list[types.Tool]:
        listed: list[types.Tool] = []
        for tool in tool_map.values():
            definition = tool.definition()
            listed.append(
                types.Tool(
                    name=definition.name,
                    description=definition.description or "",
                    inputSchema=definition.parameters,
                )
            )
        return listed

    @server.call_tool()
    async def _call_tool(tool_name: str, arguments: dict[str, Any]) -> list[types.TextContent]:
        tool = tool_map.get(tool_name)
        if tool is None:
            return [types.TextContent(type="text", text=f"Unknown tool: {tool_name!r}")]
        result = await tool.run(**(arguments or {}))
        return [types.TextContent(type="text", text=result_to_text(result))]

    return server

serve_stdio async

serve_stdio(server: Server) -> None

Run server over stdio until the client disconnects.

This is the transport Claude Desktop / Claude Code launch: the host spawns the process and speaks JSON-RPC over its stdin/stdout.

Source code in src/lazytools/mcp_server/server.py

async def serve_stdio(server: Server) -> None:
    """Run ``server`` over stdio until the client disconnects.

    This is the transport Claude Desktop / Claude Code launch: the host
    spawns the process and speaks JSON-RPC over its stdin/stdout.
    """
    from mcp.server.stdio import stdio_server

    init_options = server.create_initialization_options()
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, init_options)

default_providers

default_providers(ids: list[str] | None = None, *, allow_write: bool = False) -> list[Any]

Instantiate the default providers.

ids selects a subset (validated against :data:PROVIDER_FACTORIES); None builds them all. allow_write is threaded to every factory — False (default) yields read-only providers, True yields the write-enabled shapes (the CLI's --allow-unsafe path). A factory that raises at construction time (rare — most only fail later, at as_tools()) is skipped so one missing dependency never sinks the server.

Source code in src/lazytools/mcp_server/providers.py

def default_providers(ids: list[str] | None = None, *, allow_write: bool = False) -> list[Any]:
    """Instantiate the default providers.

    ``ids`` selects a subset (validated against :data:`PROVIDER_FACTORIES`);
    ``None`` builds them all. ``allow_write`` is threaded to every factory —
    ``False`` (default) yields read-only providers, ``True`` yields the
    write-enabled shapes (the CLI's ``--allow-unsafe`` path). A factory that
    raises at construction time (rare — most only fail later, at
    ``as_tools()``) is skipped so one missing dependency never sinks the
    server.
    """
    selected = list(PROVIDER_FACTORIES) if ids is None else ids
    unknown = [i for i in selected if i not in PROVIDER_FACTORIES]
    if unknown:
        raise ValueError(
            f"Unknown provider id(s): {', '.join(unknown)}. Known: {', '.join(PROVIDER_FACTORIES)}"
        )

    providers: list[Any] = []
    for provider_id in selected:
        try:
            providers.append(PROVIDER_FACTORIES[provider_id](allow_write))
        except Exception as exc:  # construction failure is non-fatal
            import logging

            logging.getLogger("lazytools.mcp_server").warning(
                "Could not construct provider %r: %s", provider_id, exc
            )
    return providers

expand_tools

expand_tools(providers: Sequence[Any], *, read_only: bool = True, unsafe_patterns: Iterable[str] = UNSAFE_TOOL_PATTERNS) -> dict[str, Tool]

Expand providers into a name -> Tool map, isolating failures.

Each item may be a :class:lazybridge.Tool, a plain callable, an Agent, or a ToolProvider (anything with as_tools()). Providers are expanded independently: if one raises on expansion — typically an ImportError because its optional extra is not installed (e.g. RegimeTools without lazystats[regimes]) — it is skipped with a warning and the rest still load. This is what lets a bare pip install lazytoolkit[mcp] serve datahub + statistical while regimes/web light up only once their extras are present.

On a name collision the last registration wins (with a warning), mirroring build_tool_map(collision_policy="replace").

Source code in src/lazytools/mcp_server/server.py

def expand_tools(
    providers: Sequence[Any],
    *,
    read_only: bool = True,
    unsafe_patterns: Iterable[str] = UNSAFE_TOOL_PATTERNS,
) -> dict[str, Tool]:
    """Expand ``providers`` into a ``name -> Tool`` map, isolating failures.

    Each item may be a :class:`lazybridge.Tool`, a plain callable, an Agent,
    or a ``ToolProvider`` (anything with ``as_tools()``). Providers are
    expanded **independently**: if one raises on expansion — typically an
    ``ImportError`` because its optional extra is not installed (e.g.
    ``RegimeTools`` without ``lazystats[regimes]``) — it is skipped with a
    warning and the rest still load. This is what lets a bare
    ``pip install lazytoolkit[mcp]`` serve datahub + statistical while
    regimes/web light up only once their extras are present.

    On a name collision the last registration wins (with a warning),
    mirroring ``build_tool_map(collision_policy="replace")``.
    """
    patterns = tuple(unsafe_patterns)
    result: dict[str, Tool] = {}
    for provider in providers:
        label = getattr(provider, "name", None) or type(provider).__name__
        try:
            if getattr(provider, "_is_lazy_tool_provider", False):
                tools = list(provider.as_tools())
            elif isinstance(provider, Tool):
                tools = [provider]
            elif getattr(provider, "_is_lazy_agent", False):
                tools = [Tool.wrap(provider)]  # agents carry their own name
            elif callable(provider):
                # Plain function: the Tool constructor defaults the name to
                # ``func.__name__``. (``Tool.wrap`` *requires* an explicit
                # name for callables and would raise here, silently dropping
                # the function via the except below.)
                tools = [Tool(provider)]
            else:
                raise TypeError(f"cannot expose {type(provider).__name__!r} as a tool")
        except Exception as exc:
            logger.warning("Skipping tool provider %s: %s", label, exc)
            continue

        for tool in tools:
            if read_only and _is_unsafe(tool.name, patterns):
                logger.warning(
                    "read_only: dropping mutating tool %r (matched unsafe pattern). "
                    "Pass read_only=False to expose it.",
                    tool.name,
                )
                continue
            if tool.name in result:
                logger.warning("Tool name collision on %r — keeping the later registration.", tool.name)
            result[tool.name] = tool
    return result

result_to_text

result_to_text(result: Any) -> str

Serialize a tool's return value to a single text payload for MCP.

Strings pass through untouched; bytes are decoded as UTF-8; everything else is rendered as pretty JSON (pydantic model_dump_json / to_json fast-paths first), never raising.

Source code in src/lazytools/mcp_server/server.py

def result_to_text(result: Any) -> str:
    """Serialize a tool's return value to a single text payload for MCP.

    Strings pass through untouched; bytes are decoded as UTF-8; everything
    else is rendered as pretty JSON (pydantic ``model_dump_json`` /
    ``to_json`` fast-paths first), never raising.
    """
    if result is None:
        return "null"
    if isinstance(result, str):
        return result
    if isinstance(result, (bytes, bytearray)):
        try:
            return bytes(result).decode("utf-8")
        except Exception:
            return repr(bytes(result))

    dump_json = getattr(result, "model_dump_json", None)
    if callable(dump_json):
        try:
            return dump_json()
        except Exception:
            pass

    to_json = getattr(result, "to_json", None)
    if callable(to_json):
        try:
            value = to_json()
            return value if isinstance(value, str) else json.dumps(value, default=_json_default, ensure_ascii=False)
        except Exception:
            pass

    try:
        return json.dumps(result, default=_json_default, ensure_ascii=False, indent=2)
    except Exception:
        return str(result)