Wrap a legacy class

Turn an existing service class into tools without editing it: its methods, its blocking sync calls and its own exceptions. Functions, partials, callable objects and the per-argument options (fixed=, hide=, rename=, result=) are on Wrap existing code.

flowchart LR
  L["legacy class<br/>(unchanged)"] -->|"registry.wrap(obj.method, ...)"| R[("Registry")]
  S["thin shim<br/>legacy error → ToolError"] -->|"registry.wrap(shim)"| R
  R --> X["Executor"]
  X -->|"sync method"| TH["worker thread<br/>asyncio.to_thread"]

1. The existing code

A sync service with blocking calls and its own exceptions. None of it changes.

class InvoiceService:
    """Existing code: sync, blocking, raises its own exceptions. None of it changes."""

    def __init__(self) -> None:
        self.rows: dict[str, dict[str, Any]] = {
            "INV-1": {"amount": 1200.0, "status": "open"},
            "INV-2": {"amount": 300.0, "status": "open"},
        }

    def find(self, invoice_id: str) -> dict[str, Any]:
        """Return one invoice.

        Args:
            invoice_id: Invoice number, e.g. INV-1.
        """
        time.sleep(0.01)  # a blocking database call
        return self.rows[invoice_id]  # KeyError when unknown

    def total_open(self) -> float:
        """Return the sum of all open invoices."""
        return float(sum(row["amount"] for row in self.rows.values() if row["status"] == "open"))

    def void(self, invoice_id: str, reason: str) -> str:
        """Void an invoice.

        Args:
            invoice_id: Invoice number.
            reason: Why it is voided.
        """
        self.rows[invoice_id]["status"] = "void"
        return f"{invoice_id} voided: {reason}"

2. Register its methods

Register bound methods (service.find), not the functions in the class body: self has no annotation, so build_spec raises ToolDefinitionError and points you to wrap(obj.method).

# Register bound methods. Reads may run side by side; the write needs approval.
for method in (service.find, service.total_open):
    registry.wrap(
        method,
        name=f"invoice_{method.__name__}",
        kind="db",
        effects="read_only",
        parallel_safe=True,
        timeout=5,  # a sync tool keeps running in its thread after a timeout
    )
registry.wrap(service.void, name="invoice_void", confirm=True, effects="destructive")

Choice	Why
name=f"invoice_{...}"	method names like find clash across services; the name must match [A-Za-z_][A-Za-z0-9_]*, max 64
effects="read_only", parallel_safe=True	reads may run side by side; MCP clients see a read-only hint
confirm=True, effects="destructive"	the write asks the approval port first; no port means denied
kind="db"	lets RunLimits(per_kind={"db": 4}) cap concurrent database calls per run
timeout=5	each attempt is timed; a sync tool keeps running in its thread after a timeout

The docstring becomes the description and its Args: section the parameter descriptions. A method without a docstring needs description=.

3. Blocking sync code

Sync functions and methods run in a worker thread (asyncio.to_thread), with contextvars copied, so they never block the event loop. Rules for legacy sync code:

Rule	Reason
Give it a timeout	a stuck call returns ToolTimeout to the caller; the thread itself cannot be killed
Use retries= only when the call is safe to repeat	a retry runs the method again (idempotent=True only informs MCP clients and the HTTP catalogue)
Set parallel_safe=False (the default) for code that is not thread-safe	it takes the run's exclusive lock: nothing else in the run runs meanwhile
Keep loop-bound clients out of sync code	sync code has no loop; async clients belong in a LoopLocal (Runtime)

4. Legacy exceptions

A foreign exception (here KeyError) becomes ToolFailed: kind tool-failed, with no exception text, so nothing internal leaks to the model. The executor logs the exception type at ERROR. To give the model a useful message, add a thin shim that raises ToolError; its message is shown to the model.

def find_or_explain(invoice_id: str) -> dict[str, Any]:
    """Return one invoice, or tell the model it does not exist.

    Args:
        invoice_id: Invoice number, e.g. INV-1.
    """
    try:
        return service.find(invoice_id)
    except KeyError:
        raise ToolError(f"no invoice {invoice_id}; ask the user for the number") from None


registry.wrap(find_or_explain, name="invoice_lookup", effects="read_only")

5. Run it

reviewer = CallbackApproval(lambda request: request.arguments["reason"] != "")
executor = Executor(registry, approval=reviewer)


async def main() -> None:
    print(registry.names())
    print(await executor.run("invoice_find", {"invoice_id": "INV-1"}))
    raw = await executor.run_call(ToolCall(name="invoice_find", arguments={"invoice_id": "X"}))
    print(raw.status, raw.error.kind if raw.error else "", raw.error.message if raw.error else "")
    told = await executor.run_call(ToolCall(name="invoice_lookup", arguments={"invoice_id": "X"}))
    print(told.error.kind if told.error else "", told.error.message if told.error else "")
    print(await executor.run("invoice_void", {"invoice_id": "INV-1", "reason": "duplicate"}))
    print(await executor.run("invoice_total_open"))

Output:

['invoice_find', 'invoice_total_open', 'invoice_void', 'invoice_lookup']
{'amount': 1200.0, 'status': 'open'}
error tool-failed tool 'invoice_find' failed after 1 attempt
tool-error no invoice X; ask the user for the number
INV-1 voided: duplicate
300.0

The CallbackApproval reviewer approves a void only when a reason is given. In production use a deferred review.

Other shapes

Legacy shape	Do
A function, a functools.partial, a callable object	Wrap existing code
A REST service with an OpenAPI spec	import_openapi(registry, spec, base_url=...) (OpenAPI)
LangChain tools you already have	import_langchain(registry, tools) (LangChain)
Tools on an MCP server	import_mcp(...) (MCP)

Files

File	Does
examples/howto_wrap_legacy.py	the legacy service, its registration, the shim and the calls
tests/docs/test_howto_examples.py	runs it offline and checks the output