Langclaw

langclaw.Langclaw(config=None, *, system_prompt=None, context_schema=None, enable_interpreter=False, backend=None)

Central application object for building multi-channel agent systems.

Wraps :func:~langclaw.agents.builder.create_claw_agent and the gateway infrastructure, exposing a declarative API for tool/role/channel/middleware registration.

Parameters:

Name	Type	Description	Default
config	LangclawConfig | None	Pre-built configuration. When None, loaded from env vars, .env, and ~/.langclaw/config.json via :func:~langclaw.config.schema.load_config.	None
system_prompt	str | None	Additional instructions appended after the base AGENTS.md prompt. Use this to give your app a distinct personality, domain focus, or behavioural rules without replacing the built-in defaults (memory protocol, tone, tool-use guidelines). The base ``AGENTS.md`` is always loaded first from the workspace (``~/.langclaw/workspace/AGENTS.md``). Your ``system_prompt`` is concatenated after it, separated by a blank line. To fully replace the base prompt, edit ``AGENTS.md`` directly instead. Example:: app = Langclaw( system_prompt=( "## Research Assistant\n" "You are a financial research assistant.\n" "Always check stock prices before answering." ), )	None
context_schema	type[LangclawContext] | None	Custom context schema to use for the agent. If omitted, uses the default LangclawContext.	None
enable_interpreter	bool	Opt into the sandboxed code interpreter (RLM) programmatically — equivalent to setting interpreter.enabled=true in config. Off by default. Requires the interpreter extra (uv add 'langclaw[interpreter]').	False

Source code in langclaw/app.py

def __init__(
    self,
    config: LangclawConfig | None = None,
    *,
    system_prompt: str | None = None,
    context_schema: type[LangclawContext] | None = None,
    enable_interpreter: bool = False,
    backend: Any | None = None,
) -> None:
    self._config = config or load_config()
    self._system_prompt = system_prompt
    self._context_schema = context_schema
    self._enable_interpreter = enable_interpreter
    # Optional explicit deepagents backend (instance or runtime factory)
    # applied to every agent. ``None`` (default) builds the config-selected
    # backend per agent, correctly re-rooted at each agent's workspace. Pass
    # an instance only for the advanced backends config can't express
    # (StoreBackend with a custom store/namespace, CompositeBackend, a
    # sandbox); note a shared instance is *not* re-rooted per named agent —
    # pass a ``Callable[[ToolRuntime], BackendProtocol]`` if you need that.
    self._backend = backend
    self._extra_tools: list[Any] = []
    self._extra_channels: list[BaseChannel] = []
    self._extra_middleware: list[Any] = []
    self._extra_roles: dict[str, list[str]] = {}
    self._extra_role_subagents: dict[str, list[str]] = {}
    self._extra_role_workflows: dict[str, list[str]] = {}
    self._extra_commands: list[tuple[str, Callable[[CommandContext], Awaitable[str]], str]] = []
    self._subagents: list[dict[str, Any]] = []
    self._named_agents: dict[str, dict[str, Any]] = {}
    self._workflows = WorkflowRegistry()
    self._workflow_runtime: WorkflowRuntime | None = None
    # Set at startup when ``workflows.durable_steps`` is on, else None.
    self._step_store: StepStore | None = None
    self._run_store: RunStore | None = None
    self._script_store: ScriptStore | None = None
    # File-backed store for runtime-authored (agent-written) workflow files.
    self._saved_store: SavedWorkflowStore | None = None
    self._startup_hooks: list[Callable] = []
    self._shutdown_hooks: list[Callable] = []
    self._bus: BaseMessageBus | None = None
    # Probe mode: force a WebSocket-only, isolated channel set (set by run()).
    self._probe_ws_only: bool = False
    self._probe_port: int | None = None
    self._context_defaults: dict[str, Any] = {}
    self._context_factory: (
        Callable[[InboundMessage, dict[str, Any]], Awaitable[LangclawContext]] | None
    ) = None

tool(*, roles=None)

Decorator to register a function as a LangChain tool.

If the decorated function is not already a BaseTool, it is wrapped with langchain_core.tools.tool.

Parameters:

Name	Type	Description	Default
roles	list[str] | None	Optional list of role names that should be granted access to this tool. When provided, the corresponding roles are created/updated in the RBAC config.	None

Returns:

Type	Description
Callable	A decorator that registers the tool and returns it.

Source code in langclaw/app.py

def tool(self, *, roles: list[str] | None = None) -> Callable:
    """Decorator to register a function as a LangChain tool.

    If the decorated function is not already a ``BaseTool``, it is
    wrapped with ``langchain_core.tools.tool``.

    Args:
        roles: Optional list of role names that should be granted
               access to this tool.  When provided, the corresponding
               roles are created/updated in the RBAC config.

    Returns:
        A decorator that registers the tool and returns it.
    """

    def decorator(fn: Callable) -> Any:
        from langchain_core.tools import BaseTool as _BaseTool
        from langchain_core.tools import tool as lc_tool

        t = fn if isinstance(fn, _BaseTool) else lc_tool(fn)
        check_tool_name_allowed(t.name)
        self._extra_tools.append(t)

        if roles:
            for role_name in roles:
                self._extra_roles.setdefault(role_name, []).append(t.name)

        return t

    return decorator

command(name, *, description='')

Decorator to register a custom bot command.

Commands bypass the LLM and message bus — they are fast system operations handled directly by the :class:CommandRouter.

The decorated function must accept a single :class:~langclaw.gateway.commands.CommandContext argument and return a str response.

Parameters:

Name	Type	Description	Default
name	str	Command name without the leading / (e.g. "ping").	required
description	str	Short help text shown by /help.	''

Returns:

Type	Description
Callable	A decorator that registers the command and returns the
Callable	original function.

Example::

@app.command("ping", description="check if bot is alive")
async def ping(ctx: CommandContext) -> str:
    return "Pong!"

Source code in langclaw/app.py

def command(
    self,
    name: str,
    *,
    description: str = "",
) -> Callable:
    """Decorator to register a custom bot command.

    Commands bypass the LLM and message bus — they are fast system
    operations handled directly by the :class:`CommandRouter`.

    The decorated function must accept a single
    :class:`~langclaw.gateway.commands.CommandContext` argument and
    return a ``str`` response.

    Args:
        name:        Command name without the leading ``/``
                     (e.g. ``"ping"``).
        description: Short help text shown by ``/help``.

    Returns:
        A decorator that registers the command and returns the
        original function.

    Example::

        @app.command("ping", description="check if bot is alive")
        async def ping(ctx: CommandContext) -> str:
            return "Pong!"
    """

    check_command_name_allowed(name)

    def decorator(
        fn: Callable[[CommandContext], Awaitable[str]],
    ) -> Callable[[CommandContext], Awaitable[str]]:
        self._extra_commands.append((name, fn, description))
        return fn

    return decorator

role(name, *, tools=None, subagents=None, workflows=None)

Define or update a permission role.

If the role already exists (from config or a prior call), each axis is merged independently (order-stable dedupe). Registering any role automatically enables the permissions system.

Three independent RBAC axes:

• tools — default-deny for unknown roles; ["*"] grants all.
• subagents — default-deny; subagent types reachable via the task tool. ["*"] allows every registered one.
• workflows — default-deny; workflows reachable as the workflow_<name> tool. ["*"] allows all.

Parameters:

Name	Type	Description	Default
name	str	Role identifier (e.g. "admin", "viewer").	required
tools	list[str] | None	Tool names this role may invoke. Use ["*"] for all.	None
subagents	list[str] | None	Subagent types this role may delegate to.	None
workflows	list[str] | None	Workflow names this role may invoke.	None

Source code in langclaw/app.py

def role(
    self,
    name: str,
    *,
    tools: list[str] | None = None,
    subagents: list[str] | None = None,
    workflows: list[str] | None = None,
) -> None:
    """Define or update a permission role.

    If the role already exists (from config or a prior call), each
    axis is merged independently (order-stable dedupe).  Registering
    any role automatically enables the permissions system.

    Three independent RBAC axes:

    - ``tools``     — **default-deny** for unknown roles; ``["*"]`` grants all.
    - ``subagents`` — **default-deny**; subagent types reachable via the
                      ``task`` tool. ``["*"]`` allows every registered one.
    - ``workflows`` — **default-deny**; workflows reachable as the
                      ``workflow_<name>`` tool. ``["*"]`` allows all.

    Args:
        name:      Role identifier (e.g. ``"admin"``, ``"viewer"``).
        tools:     Tool names this role may invoke. Use ``["*"]`` for all.
        subagents: Subagent types this role may delegate to.
        workflows: Workflow names this role may invoke.
    """

    def _merge(store: dict[str, list[str]], values: list[str] | None) -> None:
        existing = store.get(name, [])
        store[name] = list(dict.fromkeys(existing + (values or [])))

    # Always key the role into _extra_roles (even with no tools) so a
    # workflow-only / subagent-only role still triggers the permissions
    # merge in _effective_config.
    _merge(self._extra_roles, tools or [])
    if subagents is not None:
        _merge(self._extra_role_subagents, subagents)
    if workflows is not None:
        _merge(self._extra_role_workflows, workflows)

agent(name, *, description, display_name=None, system_prompt=None, tools=None, model=None)

Register a named agent that users can switch to via /switch <name>.

Named agents are fully independent agent instances built with the same :func:~langclaw.agents.builder.create_claw_agent factory as the main agent. Each named agent:

• Gets its own isolated LangGraph conversation thread (context_id = "agent:<name>"), so history never bleeds across modes.
• Shares the same checkpointer backend as the main agent.
• Can use a different system prompt, tool set, or model.

Users switch between agents via the built-in /agent <name> command, and can return to the main agent with /agent default.

Parameters:

Name	Type	Description	Default
name	str	Unique identifier used with /agent <name>. Must not be "default" (reserved sentinel).	required
description	str	Short description shown by /agent with no args.	required
display_name	str | None	Optional human-facing name for this agent. Injected into the system prompt so the model knows its own name, and shown alongside the routing key in /agent listings. When None, only the registered name is used.	None
system_prompt	str | None	System prompt for this agent. When None, the base AGENTS.md prompt is used unchanged.	None
tools	list[Any] | None	Explicit list of tool instances for this agent. None inherits the config-driven built-in tools without the extra tools registered on the app.	None
model	str | BaseChatModel | None	Override the default model. Accepts "provider:model" strings or a BaseChatModel.	None

Raises:

Type	Description
ValueError	If name is "default" (reserved).

Example::

app.agent(
    "researcher",
    description="Deep research mode with web tools",
    system_prompt="You are a meticulous researcher. Always cite sources.",
    tools=[web_search, web_fetch],
    model="openai:gpt-4.1",
)

Source code in langclaw/app.py

def agent(
    self,
    name: str,
    *,
    description: str,
    display_name: str | None = None,
    system_prompt: str | None = None,
    tools: list[Any] | None = None,
    model: str | BaseChatModel | None = None,
) -> None:
    """Register a named agent that users can switch to via ``/switch <name>``.

    Named agents are fully independent agent instances built with the same
    :func:`~langclaw.agents.builder.create_claw_agent` factory as the main
    agent.  Each named agent:

    - Gets its own isolated LangGraph conversation thread
      (``context_id = "agent:<name>"``), so history never bleeds across modes.
    - Shares the same checkpointer backend as the main agent.
    - Can use a different system prompt, tool set, or model.

    Users switch between agents via the built-in ``/agent <name>`` command,
    and can return to the main agent with ``/agent default``.

    Args:
        name:          Unique identifier used with ``/agent <name>``.
                       Must not be ``"default"`` (reserved sentinel).
        description:   Short description shown by ``/agent`` with no args.
        display_name:  Optional human-facing name for this agent. Injected
                       into the system prompt so the model knows its own
                       name, and shown alongside the routing key in
                       ``/agent`` listings.  When ``None``, only the
                       registered ``name`` is used.
        system_prompt: System prompt for this agent.  When ``None``, the
                       base ``AGENTS.md`` prompt is used unchanged.
        tools:         Explicit list of tool instances for this agent.
                       ``None`` inherits the config-driven built-in tools
                       without the extra tools registered on the app.
        model:         Override the default model.  Accepts
                       ``"provider:model"`` strings or a ``BaseChatModel``.

    Raises:
        ValueError: If ``name`` is ``"default"`` (reserved).

    Example::

        app.agent(
            "researcher",
            description="Deep research mode with web tools",
            system_prompt="You are a meticulous researcher. Always cite sources.",
            tools=[web_search, web_fetch],
            model="openai:gpt-4.1",
        )
    """
    if name == "default":
        raise ValueError(
            "'default' is a reserved agent name — it refers to the main agent. "
            "Choose a different name."
        )
    self._named_agents[name] = {
        "name": name,
        "description": description,
        "display_name": display_name,
        "system_prompt": system_prompt,
        "tools": tools,
        "model": model,
    }

subagent(name, *, description, graph=None, system_prompt=None, tools=None, model=None, roles=None, output='main_agent')

Register a subagent that the main agent can delegate tasks to.

Subagents are invoked by the main agent via the task tool provided by deepagents. Each subagent runs in an isolated context and returns a single result.

There are three ways to define what the subagent does:

1. Declarative — pass system_prompt (and optionally tools, model). Langclaw builds the agent, resolves tool names, and injects its middleware.

2. Pre-built graph — pass a Runnable or CompiledStateGraph via graph. Langclaw wraps it into a deepagents CompiledSubAgent and passes it through as-is. The runnable's state schema must include a messages key.

3. deepagents dict — pass a SubAgent or CompiledSubAgent TypedDict via graph. For SubAgent dicts, Langclaw prepends its middleware (channel context, RBAC). CompiledSubAgent dicts (with a runnable key) are passed through unchanged.

When graph is None, system_prompt is required.

Parameters:

Name	Type	Description	Default
name	str	Unique identifier used by the main agent when calling the task tool.	required
description	str	What this subagent does. Be specific — the main agent uses this to decide when to delegate.	required
graph	Runnable | dict[str, Any] | None	A pre-built Runnable, CompiledStateGraph, or deepagents SubAgent/CompiledSubAgent dict. Mutually exclusive with system_prompt.	None
system_prompt	str | None	Instructions for the subagent (declarative mode). Required when graph is not provided.	None
tools	list[str] | None	Tool names this subagent may use (declarative mode only). Resolved at build time against all registered tools. None inherits the main agent's full tool set.	None
model	str | BaseChatModel | None	Override the main agent's model (declarative mode only). Accepts "provider:model" strings or a BaseChatModel instance.	None
roles	list[str] | None	Reserved for future RBAC scoping of which user roles may trigger this subagent.	None
output	Literal['main_agent', 'channel']	"main_agent" (default) returns the result to the main agent. "channel" publishes the result directly to the originating channel via the message bus (declarative mode only).	'main_agent'

Raises:

Type	Description
ValueError	If neither graph nor system_prompt is provided, or if both are provided, or if output is invalid.

Example::

# Declarative — Langclaw builds the agent
app.subagent(
    "researcher",
    description="Researches topics using web search",
    system_prompt="You are a thorough researcher...",
    tools=["web_search", "web_fetch"],
    model="openai:gpt-4.1",
)

# Pre-built LangGraph graph
my_graph = create_agent("openai:gpt-4.1", tools=[...])
app.subagent(
    "my-graph",
    description="Custom LangGraph pipeline",
    graph=my_graph,
)

# deepagents SubAgent dict
app.subagent(
    "analyst",
    description="Financial analyst",
    graph={
        "system_prompt": "Analyze data.",
        "tools": [my_tool],
        "model": "openai:gpt-4.1",
    },
)

Source code in langclaw/app.py

def subagent(
    self,
    name: str,
    *,
    description: str,
    graph: Runnable | dict[str, Any] | None = None,
    system_prompt: str | None = None,
    tools: list[str] | None = None,
    model: str | BaseChatModel | None = None,
    roles: list[str] | None = None,
    output: Literal["main_agent", "channel"] = "main_agent",
) -> None:
    """Register a subagent that the main agent can delegate tasks to.

    Subagents are invoked by the main agent via the ``task`` tool
    provided by deepagents.  Each subagent runs in an isolated
    context and returns a single result.

    There are three ways to define what the subagent does:

    1. **Declarative** — pass ``system_prompt`` (and optionally
       ``tools``, ``model``).  Langclaw builds the agent, resolves
       tool names, and injects its middleware.

    2. **Pre-built graph** — pass a ``Runnable`` or
       ``CompiledStateGraph`` via ``graph``.  Langclaw wraps it
       into a deepagents ``CompiledSubAgent`` and passes it through
       as-is.  The runnable's state schema **must** include a
       ``messages`` key.

    3. **deepagents dict** — pass a ``SubAgent`` or
       ``CompiledSubAgent`` TypedDict via ``graph``.  For
       ``SubAgent`` dicts, Langclaw prepends its middleware
       (channel context, RBAC).  ``CompiledSubAgent`` dicts (with
       a ``runnable`` key) are passed through unchanged.

    When ``graph`` is ``None``, ``system_prompt`` is required.

    Args:
        name:          Unique identifier used by the main agent when
                       calling the ``task`` tool.
        description:   What this subagent does.  Be specific —
                       the main agent uses this to decide when to
                       delegate.
        graph:         A pre-built ``Runnable``, ``CompiledStateGraph``,
                       or deepagents ``SubAgent``/``CompiledSubAgent``
                       dict.  Mutually exclusive with ``system_prompt``.
        system_prompt: Instructions for the subagent (declarative mode).
                       Required when ``graph`` is not provided.
        tools:         Tool **names** this subagent may use (declarative
                       mode only).  Resolved at build time against all
                       registered tools.  ``None`` inherits the main
                       agent's full tool set.
        model:         Override the main agent's model (declarative mode
                       only).  Accepts ``"provider:model"`` strings or
                       a ``BaseChatModel`` instance.
        roles:         Reserved for future RBAC scoping of which user
                       roles may trigger this subagent.
        output:        ``"main_agent"`` (default) returns the result
                       to the main agent.  ``"channel"`` publishes
                       the result directly to the originating channel
                       via the message bus (declarative mode only).

    Raises:
        ValueError: If neither ``graph`` nor ``system_prompt`` is
                    provided, or if both are provided, or if
                    ``output`` is invalid.

    Example::

        # Declarative — Langclaw builds the agent
        app.subagent(
            "researcher",
            description="Researches topics using web search",
            system_prompt="You are a thorough researcher...",
            tools=["web_search", "web_fetch"],
            model="openai:gpt-4.1",
        )

        # Pre-built LangGraph graph
        my_graph = create_agent("openai:gpt-4.1", tools=[...])
        app.subagent(
            "my-graph",
            description="Custom LangGraph pipeline",
            graph=my_graph,
        )

        # deepagents SubAgent dict
        app.subagent(
            "analyst",
            description="Financial analyst",
            graph={
                "system_prompt": "Analyze data.",
                "tools": [my_tool],
                "model": "openai:gpt-4.1",
            },
        )
    """
    from langchain_core.runnables import Runnable as _Runnable

    if graph is not None and system_prompt is not None:
        raise ValueError(
            "'graph' and 'system_prompt' are mutually exclusive. "
            "Use 'graph' to bring a pre-built agent, or "
            "'system_prompt' for Langclaw to build one."
        )

    if graph is not None:
        if isinstance(graph, _Runnable):
            self._subagents.append(
                {
                    "name": name,
                    "description": description,
                    "runnable": graph,
                }
            )
        elif isinstance(graph, dict):
            self._subagents.append({**graph, "name": name, "description": description})
        else:
            raise TypeError(f"'graph' must be a Runnable or dict, got {type(graph).__name__}")
        return

    if system_prompt is None:
        raise ValueError("Either 'graph' or 'system_prompt' is required.")

    if output not in ("main_agent", "channel"):
        raise ValueError(
            f"Invalid output mode {output!r} for subagent {name!r}. "
            "Must be 'main_agent' or 'channel'."
        )

    self._subagents.append(
        {
            "name": name,
            "description": description,
            "system_prompt": system_prompt,
            "tools": tools,
            "model": model,
            "roles": roles,
            "output": output,
        }
    )

workflow(name, *, description='', input=None, output=None, mode='python', max_steps=None, max_concurrency=8, timeout_s=None, uses_tools=None)

Register an operator-authored workflow via decorator (issue #38).

A workflow is an async def (ctx, inp) -> output that orchestrates multi-step agent work. Each step (ctx.agent / ctx.subagent / ctx.tool / ctx.parallel) round-trips through the same bus → gateway pipeline as an ordinary message, so RBAC, rate limiting, channel context, and checkpointing are inherited. Unlike the interpreter (eval), a workflow is durable, typed, named, and RBAC-gated.

Workflows are inert unless config.workflows.enabled is True.

Example::

class Brief(BaseModel):
    topic: str

@app.workflow("research", input=Brief, description="Deep research")
async def research(ctx, inp: Brief) -> str:
    ctx.phase("gather")
    facts = await ctx.parallel([
        lambda c: c.subagent("researcher", f"Find facts on {inp.topic}"),
        lambda c: c.subagent("researcher", f"Find risks of {inp.topic}"),
    ])
    ctx.phase("synthesize")
    return await ctx.agent("writer", f"Summarize: {facts}")

Parameters:

Name	Type	Description	Default
name	str	Unique workflow handle (invoked as workflow_<name>, /workflows run <name>, cron, or PTC).	required
description	str	Becomes the tool description the LLM reads to decide when to call this workflow — exactly like a @app.tool docstring. Write it as guidance ("Research a topic across several angles in parallel; prefer over ad-hoc searches when multiple perspectives help"), not a label. Omitting it falls back to a bland "Run the '<name>' workflow." that rarely beats a plain web_search or task. (The input shape is advertised separately, from the input model below.)	''
input	type | None	Optional Pydantic model validating the run input. Its fields become the tool's argument schema, so add Field(description=...) to tell the LLM what to pass.	None
output	type | None	Optional Pydantic model validating the run output.	None
mode	str	"python" (default, recommended — you author the body; reviewed, typed, testable) or "llm_authored" (Mode 2, experimental — the LLM authors the body from this contract; an escape hatch for variable, low-stakes, supervised tasks, not a peer of python).	'python'
max_steps	int | None	Per-workflow step budget (None → global default).	None
max_concurrency	int	Fan-out width for ctx.parallel.	8
timeout_s	float | None	Per-run wall-clock budget in seconds.	None
uses_tools	list[str] | None	Tool names this workflow declares it needs.	None

Raises:

Type	Description
ValueError	If the name collides with an existing workflow, tool, subagent, named agent, or command.

Source code in langclaw/app.py

def workflow(
    self,
    name: str,
    *,
    description: str = "",
    input: type | None = None,
    output: type | None = None,
    mode: str = "python",
    max_steps: int | None = None,
    max_concurrency: int = 8,
    timeout_s: float | None = None,
    uses_tools: list[str] | None = None,
) -> Callable:
    """Register an operator-authored workflow via decorator (issue #38).

    A workflow is an ``async def (ctx, inp) -> output`` that orchestrates
    multi-step agent work.  Each step (``ctx.agent`` / ``ctx.subagent`` /
    ``ctx.tool`` / ``ctx.parallel``) round-trips through the same bus →
    gateway pipeline as an ordinary message, so RBAC, rate limiting, channel
    context, and checkpointing are inherited.  Unlike the interpreter
    (``eval``), a workflow is durable, typed, named, and RBAC-gated.

    Workflows are inert unless ``config.workflows.enabled`` is ``True``.

    Example::

        class Brief(BaseModel):
            topic: str

        @app.workflow("research", input=Brief, description="Deep research")
        async def research(ctx, inp: Brief) -> str:
            ctx.phase("gather")
            facts = await ctx.parallel([
                lambda c: c.subagent("researcher", f"Find facts on {inp.topic}"),
                lambda c: c.subagent("researcher", f"Find risks of {inp.topic}"),
            ])
            ctx.phase("synthesize")
            return await ctx.agent("writer", f"Summarize: {facts}")

    Args:
        name:            Unique workflow handle (invoked as ``workflow_<name>``,
                         ``/workflows run <name>``, cron, or PTC).
        description:     Becomes the **tool description** the LLM reads to
                         decide *when* to call this workflow — exactly like a
                         ``@app.tool`` docstring. Write it as guidance ("Research
                         a topic across several angles in parallel; prefer over
                         ad-hoc searches when multiple perspectives help"), not a
                         label. Omitting it falls back to a bland
                         ``"Run the '<name>' workflow."`` that rarely beats a
                         plain ``web_search`` or ``task``. (The *input* shape is
                         advertised separately, from the ``input`` model below.)
        input:           Optional Pydantic model validating the run input. Its
                         fields become the tool's argument schema, so add
                         ``Field(description=...)`` to tell the LLM *what* to pass.
        output:          Optional Pydantic model validating the run output.
        mode:            ``"python"`` (default, recommended — you author the
                         body; reviewed, typed, testable) or ``"llm_authored"``
                         (Mode 2, **experimental** — the LLM authors the body
                         from this contract; an escape hatch for variable,
                         low-stakes, supervised tasks, not a peer of python).
        max_steps:       Per-workflow step budget (``None`` → global default).
        max_concurrency: Fan-out width for ``ctx.parallel``.
        timeout_s:       Per-run wall-clock budget in seconds.
        uses_tools:      Tool names this workflow declares it needs.

    Raises:
        ValueError: If the name collides with an existing workflow, tool,
                    subagent, named agent, or command.
    """

    def decorator(func: Callable) -> Callable:
        spec = WorkflowSpec(
            name=name,
            fn=func,
            description=description,
            input_model=input,
            output_model=output,
            mode=mode,
            max_steps=max_steps,
            max_concurrency=max_concurrency,
            timeout_s=timeout_s,
            uses_tools=list(uses_tools or []),
        )
        self._workflows.register(spec, reserved_names=self._reserved_names())
        return func

    return decorator

add_channel(channel)

Register a custom channel alongside config-driven ones.

Source code in langclaw/app.py

def add_channel(self, channel: BaseChannel) -> None:
    """Register a custom channel alongside config-driven ones."""
    self._extra_channels.append(channel)

add_middleware(middleware)

Append middleware to the end of the built-in stack.

Source code in langclaw/app.py

def add_middleware(self, middleware: Any) -> None:
    """Append middleware to the end of the built-in stack."""
    self._extra_middleware.append(middleware)

on_startup(fn)

Decorator to register an async function called on gateway startup.

Source code in langclaw/app.py

def on_startup(self, fn: Callable) -> Callable:
    """Decorator to register an async function called on gateway startup."""
    self._startup_hooks.append(fn)
    return fn

on_shutdown(fn)

Decorator to register an async function called on gateway shutdown.

Source code in langclaw/app.py

def on_shutdown(self, fn: Callable) -> Callable:
    """Decorator to register an async function called on gateway shutdown."""
    self._shutdown_hooks.append(fn)
    return fn

run(*, probe=False, probe_port=None)

Start the multi-channel gateway (blocking).

Wires up the message bus, checkpointer, channels, cron manager, and agent, then runs GatewayManager until cancelled.

Parameters:

Name	Type	Description	Default
probe	bool	When True, run a WebSocket-only gateway with every other channel disabled regardless of config. This isolates the surface for the probe harness so test traffic never reaches a real Telegram/Discord chat. Applied at the channel-assembly seam — the user's config file is never mutated.	False
probe_port	int | None	Override the WebSocket port in probe mode (defaults to the configured channels.websocket.port).	None

Source code in langclaw/app.py

def run(self, *, probe: bool = False, probe_port: int | None = None) -> None:
    """Start the multi-channel gateway (blocking).

    Wires up the message bus, checkpointer, channels, cron manager,
    and agent, then runs ``GatewayManager`` until cancelled.

    Args:
        probe: When True, run a **WebSocket-only** gateway with every other
            channel disabled regardless of config. This isolates the surface
            for the probe harness so test traffic never reaches a real
            Telegram/Discord chat. Applied at the channel-assembly seam — the
            user's config file is never mutated.
        probe_port: Override the WebSocket port in probe mode (defaults to the
            configured ``channels.websocket.port``).
    """
    self._probe_ws_only = probe
    self._probe_port = probe_port
    asyncio.run(self._run_async())