Tools

octomate.capabilities.gateway

Gateway capability: an agent's routing spellbook.

The spells decide where a turn goes, who handles it, and what it is about. Each is opaque on its own, so the instruction opens with plain words for what they actually do:

• inspect: reveal agent routes, suggested addresses, or projects.
• summon: hand the conversation to another agent (a handoff — they take over from a brief). The graph reads the recorded decision after the run.
• teleport: continue the same agent in a new thread, carrying the history forward. Deferred, so the graph can fork the history and resume. With a project, the place is that project's workspace — the door out of a throwaway tree into one whose work is kept — and new_thread=false binds the current thread.
• dismiss: give a project thread's workspace back once the work in it is done. Recorded, and performed by the graph when the turn ends and its work is saved.
• commission: draw another agent into working a self-contained task in the background and return its report — an ordinary awaited tool call, never a deferral; the caller keeps the conversation and the user sees none of it.
• whisper: a quiet follow-up to an accomplice by name; it keeps its own context.
• send: deliver content mid-run without ending the turn, here or in the asking user's direct messages. It lives here rather than in its own capability because naming somewhere other than "here" is a routing decision, and this is where the channel registry and the run's own address already are.

The policy itself — what each spell may name, and the decision it records — lives on octomate.managers.gateway.OctomateSession; this capability is Inkling's translation of it into pydantic-ai: refusals become ModelRetry, a teleport becomes a deferral, and a send rides the run event stream. The accomplice spells stay wholly here — they spawn pydantic-ai runs, which no other runtime's gateway offers.

COMMISSION_TIMEOUT module-attribute

COMMISSION_TIMEOUT = 900.0

GATEWAY_INSTRUCTION_TEMPLATE module-attribute

GATEWAY_INSTRUCTION_TEMPLATE = "## Gateway — decide where this conversation goes and who handles it\n\nThese tools route the conversation. Default to handling it yourself: if you can answer\nwell or do the work, do it and call none of them. Routing is the exception — reach for a\ntool only when one of the signals below clearly fires.\n\n### Addresses and discovery\nFor `{teleport}`, `destination` is a `ChannelAddress` object, or\n`destination=null` to use the current conversation's address. An address carries\n`channel_tentacle_id`, `chat_type`, `chat_id`, `user_id`, and optional\n`channel_thread_id` and `shared` visibility. Use known identifiers and preserve\nvisibility; `user_id` must match the requesting user's linked identity on that\nchannel. When no current conversation is attached, supply an explicit address.\n\n`{inspect}` with `reveal=\"destinations\"` suggests addresses without creating anything.\nTo look further, add `channel` to browse one connected channel a level at a time:\nan address with `metadata.inside` is a place to open by passing that value as\n`inside`, and one without it can host a new thread unless `metadata.barred` says\nwhy not. A channel that cannot be browsed says so.\nYou can also supply a known address directly; discovery is optional, and each\noperation validates the address independently. Suggestions are not an exhaustive\nlist or a guarantee that every spell can use each address.\n`new_thread=true` is the default: create a thread at the address. For creation,\nuse `channel_thread_id=null`; an existing thread cannot contain another thread.\n\n### `{summon}` — hand off to another agent\nSummon transfers the conversation to a specialist who takes over this turn *and its\nfollow-ups*: a real, sticky handoff, so the bar is high. Summon only when:\n- The request needs a capability you lack — e.g. running or editing code in a real\n  repository or environment, or a domain another agent is described for.\n- It is substantial specialist work another agent would do markedly better, not\n  something you can handle from what you already know.\n\nDo NOT summon when:\n- You can already answer or do it — length or a technical-sounding topic is not a reason.\n- You are only mildly unsure — ask the user a clarifying question instead.\n- No route clearly fits — handle it yourself or ask; never summon on a guess.\n\nWhen one fires, call `{inspect}` with `reveal=\"routes\"` first to see the agents and what\neach is for. Every route carries a claim: its ability (what that agent+model is for)\nand the effort levels it accepts — pick the route whose ability covers the work. Set\n`effort` only when the user explicitly asked for a level; otherwise leave it unset so\nthe agent's own default applies. Then `{summon}` — copying its `agent_id` and `model`\nexactly from that route, and writing a self-contained brief since the other agent may\nnot see this chat. Summon hands over this conversation where it is; to continue\nsomewhere else, `{teleport}` there with `resume=true` and summon from the thread you\nland in. It is refused on a group's main conversation and in native sessions. You\nyourself are not a valid summon target.\n\n### `{teleport}` — relocate yourself\nMove this conversation into a new thread that *you* keep handling, carrying\neverything said so far. Use it for multi-step or long-running work that deserves its\nown thread but that you are the right one to do — no other agent involved.\nOmit `destination` or use null for the current address, or supply another address.\nTeleport requires attached conversation history that the agent can copy, and the\ndestination must support continuing that agent. Shared history can only move into\na sub-thread of the current chat; moving to another address requires a private\nsource. If a requested new thread cannot be created, the move fails without\nsilently continuing at its parent.\n\nTo work on a project, add `project` (from `{inspect}` with `reveal=\"projects\"`), and\n`ref` — a branch, tag or commit — only when the default branch is the wrong place to\nstart: the thread you land in is bound to it and you resume in its workspace, where\nwork is kept. A thread about no project runs in a throwaway tree, so do this before\nyou start work, not after. From inside a thread, use `destination=null`,\n`new_thread=false` and `project` to bind this one without opening another.\nStaying put requires a project; a thread binds once, and a different project is\na different thread. Native sessions must move into a new thread.\n\n### `{scheme}` — take it to the user privately\nContinue one-to-one with the person who asked, in their direct messages: for work that\nis theirs alone, or that does not belong in front of the group. Whoever already handles\ntheir direct messages picks it up, so write `brief` self-contained — it may not be you,\nand they cannot see this chat. `hint` opens the conversation over there and is the\nfirst thing they read; nothing is posted here, so close out your own reply by saying\nthe work is moving, not what it is. Its destination is a target object with\n`kind=\"dm\"`, optionally naming a connected `channel` ID for the user's linked\naccount there. Omitting the destination uses this channel's DM; that local move\nrequires a shared source. Another connected channel's DM can also be selected\nfrom a private source. The tool refuses unsupported or unlinked destinations.\n\n### `{dismiss}` — give the workspace back when the work is done\nA thread about a project keeps its workspace between turns, which is what makes its\nnext turn a resume. When the work is finished for good — merged, delivered, or\ndropped — say so with `{dismiss}`: the tree is released when this turn ends, after\nthis turn's work is saved to the project's mirror, and a later message here forks it\nafresh from there, so nothing is lost but the disk. Not for a pause — a thread\nwaiting on someone keeps its tree, and idle ones are swept on their own. Wrap up in\nthe same turn; your working directory goes with it.\n"

HANDOFF_INSTRUCTION module-attribute

HANDOFF_INSTRUCTION = "\n\n### Writing a brief\nA `summon` or `scheme` brief is the receiver's whole opening prompt, and it is\nbudgeted. Say, in this order and only where it changes what they do next: the goal\nin the user's own words; the constraints they set; the decisions made and why; what\nis done; what was tried and failed, so it is not tried again; what is left, first\nitem first; and what a finished result looks like. Leave the rest to the ledger:\nthe receiver reads every thread this person has spoken in, so cite a message by\nits `#msg:<id>` handle and name what to search for instead of repeating it. Do\nnot restate the project — the workspace's own instructions carry that.\n"

ACCOMPLICE_INSTRUCTION module-attribute

ACCOMPLICE_INSTRUCTION = "You are an accomplice: another agent commissioned you, and your reply is\nyour report back to it. There is no user here to ask, and anything that needs a\nhuman approval is declined immediately — work from the brief you were given within\nwhat you can do unaided; if something is under-specified or unapprovable, state the\nassumption or the blocker in your report and proceed.\n"

SEND_INSTRUCTION_TEMPLATE module-attribute

SEND_INSTRUCTION_TEMPLATE = "\n### `{send}` — deliver something now, without ending your turn\nFor a progress update, an intermediate result, or an image/file you produced along\nthe way. Anything sent this way is already delivered: your final reply continues from\nthere — summarize or extend it, never restate it. If everything worth saying went out\nalready, close with a short wrap-up rather than re-sending it.\n\n`destination` defaults to a target object with `kind=\"here\"`. A target with\n`kind=\"dm\"` delivers to the person who asked, privately —\nfor something that is *for them*, like a summary sent over; say in your reply that you\nsent it. `dm` hands nothing over: you keep this conversation and nobody picks the work\nup there, so use `{scheme}` when the work itself should continue privately. Asking for\n`dm` while already in that person's direct messages is fine — it lands here.\nA DM target can name a connected `channel` ID to reach their linked account there.\nIf no current conversation is attached, that channel must be explicit. Use the\nselected address's `channel_tentacle_id` as `channel`; the recipient is resolved\nfrom the user's linked profile.\n\nTo thread onto a specific message in a busy chat, lead with a reply segment whose id\nis that message's `#msg:<id>` handle — it must be the first segment. To ping someone,\nuse an `at` segment with their user id.\n"

COMMISSION_INSTRUCTION module-attribute

COMMISSION_INSTRUCTION = "\n### `commission` — put another agent to work in the background (you keep the conversation)\nWhere `summon` hands the conversation away, `commission` does not: another agent works a\nself-contained task and the tool returns its report — the user sees only your reply.\nCall `inspect` with `reveal=\"routes\"` and no `channel` to choose a route from this\nconversation's channel; the same claim and effort rules as `summon` apply.\nGive the accomplice a short mnemonic `name`. The brief must stand alone: the\naccomplice cannot see this chat and has no user to ask, so include the goal, the\nrelevant context, and what a finished result looks like. Several commissions in one\nreply run concurrently.\n\n### `whisper` — a quiet word to an accomplice\nContinue an accomplice's work by `name`: it remembers everything it did. Use it to\nrefine or extend that work instead of commissioning a new accomplice.\n"

Accomplices dataclass

The accomplices one stretch of a run called and has not heard back from.

Source code in octomate/capabilities/gateway.py

@dataclass
class Accomplices:
    """The accomplices one stretch of a run called and has not heard back from."""

    working: list[str] = field(default_factory=list)

    async def watch(
        self, stream: AsyncIterable[AgentStreamEvent]
    ) -> AsyncIterator[AgentStreamEvent]:
        """`stream` with each accomplice's start before its call and its finish
        before its result. One left unanswered finishes with the stretch: cancelled
        with it, or failed."""
        try:
            async for event in stream:
                lifecycle = self.observe(event)
                if lifecycle is not None:
                    # One dynamic-boundary cast: pydantic-ai types the stream as
                    # AgentStreamEvent, consumers match the concrete octomate event.
                    yield cast(AgentStreamEvent, lifecycle)
                yield event
        except (Exception, asyncio.CancelledError) as error:
            cancelled = isinstance(error, asyncio.CancelledError)
            for settled in self.abandon("cancelled" if cancelled else "failed"):
                yield settled
            raise
        for settled in self.abandon("failed"):
            yield settled

    def observe(
        self, event: AgentStreamEvent
    ) -> SubagentStartedEvent | SubagentSettledEvent | None:
        if isinstance(event, FunctionToolResultEvent):
            part = event.part
            if part.tool_call_id not in self.working:
                return None
            self.working.remove(part.tool_call_id)
            if isinstance(part, RetryPromptPart):
                return SubagentSettledEvent(
                    invocation_id=part.tool_call_id,
                    status="failed",
                    response=part.model_response(),
                )
            return SubagentSettledEvent(
                invocation_id=part.tool_call_id,
                status="completed" if part.outcome == "success" else "failed",
                response=part.model_response_str(),
            )
        if not isinstance(event, FunctionToolCallEvent) or event.part.tool_name not in {
            GatewayTool.COMMISSION,
            GatewayTool.WHISPER,
        }:
            return None
        name = event.part.args_as_dict().get("name")
        if not isinstance(name, str) or not name:
            # The spell refuses a nameless call, so no accomplice starts.
            return None
        self.working.append(event.part.tool_call_id)
        return SubagentStartedEvent(
            invocation_id=event.part.tool_call_id,
            kind="commission"
            if event.part.tool_name == GatewayTool.COMMISSION
            else "whisper",
            name=name,
        )

    def abandon(self, status: SubagentActivityStatus) -> list[AgentStreamEvent]:
        settled = [
            cast(
                AgentStreamEvent,
                SubagentSettledEvent(invocation_id=invocation_id, status=status),
            )
            for invocation_id in self.working
        ]
        self.working.clear()
        return settled

working class-attribute instance-attribute

working: list[str] = field(default_factory=list)

watch async

watch(
    stream: AsyncIterable[AgentStreamEvent],
) -> AsyncIterator[AgentStreamEvent]

stream with each accomplice's start before its call and its finish before its result. One left unanswered finishes with the stretch: cancelled with it, or failed.

Source code in octomate/capabilities/gateway.py

async def watch(
    self, stream: AsyncIterable[AgentStreamEvent]
) -> AsyncIterator[AgentStreamEvent]:
    """`stream` with each accomplice's start before its call and its finish
    before its result. One left unanswered finishes with the stretch: cancelled
    with it, or failed."""
    try:
        async for event in stream:
            lifecycle = self.observe(event)
            if lifecycle is not None:
                # One dynamic-boundary cast: pydantic-ai types the stream as
                # AgentStreamEvent, consumers match the concrete octomate event.
                yield cast(AgentStreamEvent, lifecycle)
            yield event
    except (Exception, asyncio.CancelledError) as error:
        cancelled = isinstance(error, asyncio.CancelledError)
        for settled in self.abandon("cancelled" if cancelled else "failed"):
            yield settled
        raise
    for settled in self.abandon("failed"):
        yield settled

observe

observe(
    event: AgentStreamEvent,
) -> SubagentStartedEvent | SubagentSettledEvent | None

Source code in octomate/capabilities/gateway.py

def observe(
    self, event: AgentStreamEvent
) -> SubagentStartedEvent | SubagentSettledEvent | None:
    if isinstance(event, FunctionToolResultEvent):
        part = event.part
        if part.tool_call_id not in self.working:
            return None
        self.working.remove(part.tool_call_id)
        if isinstance(part, RetryPromptPart):
            return SubagentSettledEvent(
                invocation_id=part.tool_call_id,
                status="failed",
                response=part.model_response(),
            )
        return SubagentSettledEvent(
            invocation_id=part.tool_call_id,
            status="completed" if part.outcome == "success" else "failed",
            response=part.model_response_str(),
        )
    if not isinstance(event, FunctionToolCallEvent) or event.part.tool_name not in {
        GatewayTool.COMMISSION,
        GatewayTool.WHISPER,
    }:
        return None
    name = event.part.args_as_dict().get("name")
    if not isinstance(name, str) or not name:
        # The spell refuses a nameless call, so no accomplice starts.
        return None
    self.working.append(event.part.tool_call_id)
    return SubagentStartedEvent(
        invocation_id=event.part.tool_call_id,
        kind="commission"
        if event.part.tool_name == GatewayTool.COMMISSION
        else "whisper",
        name=name,
    )

abandon

abandon(
    status: SubagentActivityStatus,
) -> list[AgentStreamEvent]

Source code in octomate/capabilities/gateway.py

def abandon(self, status: SubagentActivityStatus) -> list[AgentStreamEvent]:
    settled = [
        cast(
            AgentStreamEvent,
            SubagentSettledEvent(invocation_id=invocation_id, status=status),
        )
        for invocation_id in self.working
    ]
    self.working.clear()
    return settled

GatewayCapability dataclass

Bases: AbstractCapability[None]

The gateway spells as one turn's toolset: inspect, summon, teleport, scheme, send and dismiss, plus the accomplice spells when there is a thread to run them in.

Source code in octomate/capabilities/gateway.py

@dataclass
class GatewayCapability(AbstractCapability[None]):
    """The gateway spells as one turn's toolset: inspect, summon, teleport, scheme,
    send and dismiss, plus the accomplice spells when there is a thread to run
    them in."""

    # The turn's policy and decision slot, shared with the graph that acts on it.
    session: OctomateSession
    # What running an accomplice takes, beyond the session's own deps. None on a
    # gateway built only to route, which then does not offer the accomplice spells.
    conversations: ConversationManager | None = None
    commission_timeout: float = COMMISSION_TIMEOUT
    commissioning: bool = field(default=False, init=False)
    toolset: FunctionToolset[None] | None = field(default=None, init=False, repr=False)

    @property
    def decision(self) -> GatewayDecision | None:
        return self.session.decision

    def __post_init__(self) -> None:
        # Nothing built from runtime state may reach a tool definition: they are a
        # provider prompt-cache breakpoint (`anthropic_cache_tool_definitions`) at the
        # front of the prefix, so a schema that varies forks it into variants that never
        # warm each other. Hence plain `str` routes, validated by `claimed_route`
        # against the list `inspect` returns — a tool *result*, after the breakpoint.
        toolset: FunctionToolset[None] = FunctionToolset(id=GATEWAY_TOOLSET_ID)
        toolset.tool(name=GatewayTool.INSPECT)(self.inspect)
        toolset.tool(name=GatewayTool.SUMMON, retries=2)(self.summon)
        # `retries` to match its siblings: teleport refuses a surface with no
        # sub-thread to open, so it needs the same room to be told and correct.
        toolset.tool(name=GatewayTool.TELEPORT, retries=2, requires_approval=True)(
            self.teleport
        )
        toolset.tool(name=GatewayTool.SCHEME, retries=2)(self.scheme)
        toolset.tool(name=GatewayTool.SEND, retries=2)(self.send)
        toolset.tool(name=GatewayTool.DISMISS, retries=2)(self.dismiss)
        if (
            self.session.agents is not None
            and self.conversations is not None
            and self.session.thread_id is not None
            and self.session.conversation_address is not None
        ):
            self.commissioning = True
            toolset.tool(name=GatewayTool.COMMISSION, retries=2)(self.commission)
            toolset.tool(name=GatewayTool.WHISPER, retries=2)(self.whisper)
        self.toolset = toolset

    def commission_deps(
        self,
    ) -> tuple[dict[str, AgentTentacle], ConversationManager, UUID7, ChannelAddress]:
        """The live handles the accomplice spells run with. Registration only
        offers the spells when all four are set, so a miss here is a
        construction bug, not a model mistake."""
        if (
            self.session.agents is None
            or self.conversations is None
            or self.session.thread_id is None
            or self.session.conversation_address is None
        ):
            raise RuntimeError(
                "the accomplice spells need agents, conversations, a thread and an address"
            )
        return (
            self.session.agents,
            self.conversations,
            self.session.thread_id,
            self.session.conversation_address,
        )

    async def run_accomplice(
        self,
        ctx: RunContext[None],
        *,
        child: Conversation,
        run_name: str,
        prompt: str,
        model: Model | str | None,
        effort: str | None,
    ) -> str:
        agents, conversations, thread_id, conversation_address = self.commission_deps()
        if ctx.run_id is None:
            raise RuntimeError("a commission needs the parent run id for the run tree")
        agent = agents[child.agent_tentacle_id]
        try:
            # The subagent contract (non-interactive, no capabilities,
            # addressed at the pre-ensured child conversation) lives on
            # `subagent_run`; the accomplice framing is this spawner's and is
            # passed here. An accomplice that defers anyway ends its run with
            # DeferredToolRequests, surfaced loudly below instead of parking a
            # batch nothing resumes.
            result = await asyncio.wait_for(
                agent.subagent_run(
                    prompt,
                    conversation_address=conversation_address,
                    thread_id=thread_id,
                    conversation_id=child.id,
                    run_name=run_name,
                    model=model,
                    effort=effort,
                    instructions=ACCOMPLICE_INSTRUCTION,
                ),
                timeout=self.commission_timeout,
            )
        except TimeoutError:
            raise ModelRetry(
                f"The accomplice {child.subagent_id!r} exceeded "
                f"{self.commission_timeout:.0f}s and was stopped. What it "
                f"recorded is kept — `{GatewayTool.WHISPER}` to continue "
                "it, or break the work into smaller briefs."
            ) from None
        # The runner recorded its turn without knowing its place;
        # the spawner stamps the run tree after the fact.
        await conversations.link_parent_run(
            result.run_id,
            parent_run_id=ctx.run_id,
            parent_tool_call_id=ctx.tool_call_id,
        )
        output = result.output
        if isinstance(output, DeferredToolRequests):
            raise ModelRetry(
                f"The accomplice {child.subagent_id!r} tried to ask the "
                "user or defer — an accomplice has no user. Write a "
                "self-contained brief that needs no clarification, then "
                "commission another."
            )
        if isinstance(output, str):
            return output
        if isinstance(output, Iterable):
            return "\n\n".join(str(part) for part in output)
        return str(output)

    async def inspect(
        self,
        ctx: RunContext[None],
        reveal: InspectFacet,
        channel: str | None = None,
        inside: str | None = None,
    ) -> list[AgentRoute] | list[ChannelAddress] | list[ProjectSummary]:
        """Reveal one facet of what this conversation can reach.

        Args:
            reveal: `routes` — agent/model choices on the current or selected channel.
                Summon and commission use the current channel's routes.
                `destinations` — optional address suggestions, validated independently
                when used; known addresses need not appear in this list. With
                `channel`, one level of that channel's addresses instead.
                `projects` — the projects this deployment can work on, for `teleport`.
            channel: A connected channel ID from an address. For `routes`, omit it
                to inspect the current conversation's routes. For `destinations`,
                name it to browse that channel as the requesting user: an address
                whose `metadata.inside` is set is a place to open, one without it
                can host a new thread unless `metadata.barred` says why not.
            inside: For `destinations` with `channel`, the `metadata.inside` of a
                listed address, to list what that place holds. Omit it for the
                channel's top level.
        """
        try:
            return await self.session.inspect(reveal, channel, inside)
        except GatewayRefusal as refusal:
            raise ModelRetry(str(refusal)) from refusal

    async def summon(
        self,
        ctx: RunContext[None],
        agent_id: str,
        model: str,
        hint: str,
        reason: str,
        brief: Annotated[str, Field(max_length=8_000)],
        effort: str | None = None,
    ) -> str:
        """Hand this conversation to another Octomate agent, who takes it over where
        it is.

        Args:
            agent_id: The target agent, copied exactly from an `inspect` route
                (`reveal="routes"`).
            model: That route's model, copied exactly.
            hint: A short, user-facing note announcing the handoff, recorded with it.
            reason: One line on why this agent fits — recorded with the handoff, not
                shown to the user as the reply.
            brief: The self-contained brief the other agent starts from. It becomes
                their opening prompt and they cannot see this conversation, so give
                the goal, the relevant context and decisions, what's been tried, and
                what a finished result looks like. If anything you carry says how to
                hand off — a handoff skill, a plugin, your own harness's instructions
                — follow it; this is the floor. Reference rather than repeat: the
                receiver reads every thread this person has spoken in, so cite a
                message by its `#msg:<id>` handle and say what to search for instead
                of pasting it. A brief over the size budget is refused, never trimmed.
            effort: How hard the agent should think, from the effort levels the
                route's claim offers. Set it only when the user explicitly asked
                for a level; omitted, the agent's own default applies.
        """
        try:
            return await self.session.summon(
                agent_id=agent_id,
                model=model,
                hint=hint,
                reason=reason,
                brief=brief,
                effort=effort,
            )
        except GatewayRefusal as refusal:
            raise ModelRetry(str(refusal)) from refusal

    async def teleport(
        self,
        ctx: RunContext[None],
        hint: str,
        destination: ChannelAddress | None = None,
        project: str | None = None,
        ref: str | None = None,
        new_thread: bool = True,
        resume: bool = False,
    ) -> str:
        """Continue this conversation yourself somewhere else; everything said so
        far comes with you. This turn ends on it, and your context is intact
        there.

        Args:
            hint: The short, user-facing thread-starter message.
            destination: The address at which to create a thread; omit for the
                current chat. The owning channel validates it independently of
                discovery and requires the requesting user's linked identity.
                Shared history stays under its current parent. Teleport requires
                attached history that the agent can copy.
            new_thread: Create a thread before continuing. Set false only to bind
                the current thread to a project without opening another one.
                For creation, the address must have no `channel_thread_id`.
            project: A project's name, copied exactly from `inspect`
                (`reveal="projects"`), for a conversation about no project: the
                thread you land in is bound to it, and you resume in its workspace,
                where work is kept. A conversation about a project carries it
                wherever it goes, so omit this there.
            ref: The branch, tag or commit that workspace starts from; omit it for
                the project's default branch.
            resume: Carry on working there at once, as after any tool call. Leave
                it false when the move is all that was asked: you continue when
                the user next writes there.
        """
        try:
            decision = await self.session.teleport(
                hint=hint,
                destination=destination,
                new_thread=new_thread,
                project=project,
                ref=ref,
                resume=resume,
            )
        except GatewayRefusal as refusal:
            raise ModelRetry(str(refusal)) from refusal
        raise CallDeferred(metadata=decision.metadata())

    async def scheme(
        self,
        ctx: RunContext[None],
        hint: str,
        brief: Annotated[str, Field(max_length=8_000)],
        destination: SchemeTarget = DIRECT_TARGET,
    ) -> str:
        """Continue this with the user one-to-one, in their direct messages.

        Whoever already handles that user's direct messages picks the work up — you do
        not choose them, and their direct messages keep the owner they had. Use it for
        work that is that person's alone, or that does not belong in front of the group.

        Args:
            hint: The short, user-facing line that opens the conversation over there,
                where they will read it before anything else. Say that the work is
                being picked up — not the brief, which is written for whoever answers.
            brief: The self-contained brief whoever answers there starts from. They
                cannot see this conversation, so give the goal, the relevant context and
                decisions, what's been tried, and what a finished result looks like. If
                anything you carry says how to hand off — a handoff skill, a plugin,
                your own harness's instructions — follow it; this is the floor.
                Reference rather than repeat: the receiver reads every thread this
                person has spoken in, so cite a message by its `#msg:<id>` handle and
                say what to search for instead of pasting it. A brief over the size
                budget is refused, never trimmed.
            destination: A target object with `kind="dm"`. Omit `channel` for this
                channel's DM when moving from a shared source, or name another
                connected channel ID, including from a private source. The user's
                linked profile determines the recipient.
        """
        try:
            return await self.session.scheme(
                hint=hint, brief=brief, destination=destination
            )
        except GatewayRefusal as refusal:
            raise ModelRetry(str(refusal)) from refusal

    async def send(
        self,
        ctx: RunContext[None],
        segments: list[MessageSegment],
        destination: SendTarget = HERE_TARGET,
    ) -> ToolReturn[str]:
        """Deliver these segments immediately, without ending your turn — a progress
        update, an intermediate result, an image or a file. What you send here is
        already delivered: do NOT repeat it in your final reply.

        Args:
            segments: What to deliver.
            destination: A target object with `kind="here"` (the default) or
                `kind="dm"`. A DM target can name a connected `channel` ID; omit
                it for this channel. With no current conversation, name a channel
                explicitly. When choosing from an inspected address, use its
                `channel_tentacle_id` as `channel`; the linked profile supplies the
                recipient. Say in your reply when you sent it elsewhere.
        """
        try:
            address = await self.session.resolve_send(destination)
        except GatewayRefusal as refusal:
            raise ModelRetry(str(refusal)) from refusal
        return ToolReturn(
            return_value="sent",
            metadata=[MessageSentEvent(segments=segments, destination=address)],
        )

    async def dismiss(self, ctx: RunContext[None]) -> str:
        """Give this thread's workspace back, now that the work in it is done —
        merged, delivered, or dropped for good, not paused. It is released when
        this turn ends, after this turn's work is saved to the project's mirror,
        so nothing is lost: a later message on this thread forks it afresh from
        there. Only the disk goes, and it goes now rather than after the idle
        window the sweep would otherwise wait out.
        """
        try:
            return await self.session.dismiss()
        except GatewayRefusal as refusal:
            raise ModelRetry(str(refusal)) from refusal

    async def commission(
        self,
        ctx: RunContext[None],
        name: str,
        agent_id: str,
        model: str,
        brief: str,
        effort: str | None = None,
    ) -> str:
        """Draw another Octomate agent into working a task and get its report
        back — the user sees none of it.

        Args:
            name: Your name for this accomplice — short and mnemonic, e.g.
                `repo-audit`. `whisper` to it later to follow up.
            agent_id: The agent to draw in, copied exactly from an
                `inspect` route for this conversation's channel (omit `channel`).
            model: That route's model, copied exactly.
            brief: The self-contained work order. The accomplice cannot see
                this conversation and has no user to ask, so give the
                goal, the relevant context and decisions, and what a
                finished result looks like.
            effort: How hard the accomplice should think, from the levels the
                route's claim offers. Set it only when the user
                explicitly asked for a level.
        """
        agents, conversations, thread_id, _ = self.commission_deps()
        if not name.strip():
            raise ModelRetry("Give the accomplice a short, mnemonic name.")
        if agent_id == self.session.current_agent_id:
            raise ModelRetry(
                f"Cannot commission yourself {self.session.current_agent_id!r}. "
                f'Call `{GatewayTool.INSPECT}` with `reveal="routes"` to choose a valid route.'
            )
        try:
            route = self.session.claimed_route(
                agent_id, model, effort, spell="commission"
            )
        except GatewayRefusal as refusal:
            raise ModelRetry(str(refusal)) from refusal
        run_model = agents[agent_id].models.get(route.model)
        if run_model is None:
            raise ModelRetry(
                f"Agent {agent_id!r} does not serve model {model!r}. "
                f'Call `{GatewayTool.INSPECT}` with `reveal="routes"` and copy a route exactly.'
            )
        # The calling run's own conversation is the parent — the react
        # graph put its id on the RunContext. No id means the gate is
        # mounted outside a live run: raise, never conjure a parent.
        if ctx.conversation_id is None:
            raise RuntimeError("a commission needs the calling run's conversation id")
        parent_id = uuid.UUID(ctx.conversation_id)
        child = await conversations.ensure(
            thread_id,
            agent_tentacle_id=agent_id,
            subagent_id=name,
            parent_conversation_id=parent_id,
        )
        if child.parent_conversation_id != parent_id or child.runs:
            raise ModelRetry(
                f"{name!r} is already at work — `{GatewayTool.WHISPER}` "
                "to it, or pick a new name."
            )
        return await self.run_accomplice(
            ctx,
            child=child,
            run_name="commission",
            prompt=brief,
            model=run_model,
            effort=effort,
        )

    async def whisper(self, ctx: RunContext[None], name: str, message: str) -> str:
        """A quiet word to an accomplice you commissioned. It keeps
        everything it did — your whisper continues its work, not a fresh
        start.

        Args:
            name: The name you gave the accomplice when you commissioned it.
            message: The follow-up work order — self-contained, like a
                brief; the accomplice still cannot see this conversation.
        """
        _, conversations, thread_id, _ = self.commission_deps()
        if ctx.conversation_id is None:
            raise RuntimeError("a whisper needs the calling run's conversation id")
        parent_id = uuid.UUID(ctx.conversation_id)
        hands = await conversations.subagents(parent_id)
        stored = next((hand for hand in hands if hand.subagent_id == name), None)
        if stored is None:
            live = ", ".join(sorted(hand.subagent_id for hand in hands))
            raise ModelRetry(
                f"No accomplice named {name!r}. "
                + (
                    f"Accomplices at work: {live}."
                    if live
                    else "No accomplices are at work."
                )
                + f" `{GatewayTool.COMMISSION}` one to start."
            )
        child = await conversations.ensure(
            thread_id,
            agent_tentacle_id=stored.agent_tentacle_id,
            subagent_id=name,
            parent_conversation_id=parent_id,
        )
        return await self.run_accomplice(
            ctx,
            child=child,
            run_name="whisper",
            prompt=message,
            model=None,
            effort=None,
        )

    def get_instructions(self) -> str:
        instructions = gateway_instructions(lambda name: name) + HANDOFF_INSTRUCTION
        if self.commissioning:
            return instructions + COMMISSION_INSTRUCTION
        return instructions

    async def wrap_run_event_stream(
        self,
        ctx: RunContext[None],
        *,
        stream: AsyncIterable[AgentStreamEvent],
    ) -> AsyncIterable[AgentStreamEvent]:
        """Put each `send`'s stashed `MessageSentEvent` on the run stream,
        where the consumer rendering this run delivers it, and each accomplice's
        start and finish around its call."""
        async for event in Accomplices().watch(stream):
            yield event
            if (
                isinstance(event, FunctionToolResultEvent)
                and isinstance(event.part, ToolReturnPart)
                and event.part.tool_name == GatewayTool.SEND
                and isinstance(event.part.metadata, list)
            ):
                for sent_event in event.part.metadata:
                    # One dynamic-boundary cast: pydantic-ai types the stream as
                    # AgentStreamEvent, consumers match the concrete octomate event.
                    yield cast(AgentStreamEvent, sent_event)

    def get_toolset(self) -> AbstractToolset[None] | None:
        return self.toolset

session instance-attribute

session: OctomateSession

conversations class-attribute instance-attribute

conversations: ConversationManager | None = None

commission_timeout class-attribute instance-attribute

commission_timeout: float = COMMISSION_TIMEOUT

commissioning class-attribute instance-attribute

commissioning: bool = field(default=False, init=False)

toolset class-attribute instance-attribute

toolset: FunctionToolset[None] | None = field(
    default=None, init=False, repr=False
)

decision property

decision: GatewayDecision | None

commission_deps

commission_deps() -> tuple[
    dict[str, AgentTentacle],
    ConversationManager,
    UUID7,
    ChannelAddress,
]

The live handles the accomplice spells run with. Registration only offers the spells when all four are set, so a miss here is a construction bug, not a model mistake.

Source code in octomate/capabilities/gateway.py

def commission_deps(
    self,
) -> tuple[dict[str, AgentTentacle], ConversationManager, UUID7, ChannelAddress]:
    """The live handles the accomplice spells run with. Registration only
    offers the spells when all four are set, so a miss here is a
    construction bug, not a model mistake."""
    if (
        self.session.agents is None
        or self.conversations is None
        or self.session.thread_id is None
        or self.session.conversation_address is None
    ):
        raise RuntimeError(
            "the accomplice spells need agents, conversations, a thread and an address"
        )
    return (
        self.session.agents,
        self.conversations,
        self.session.thread_id,
        self.session.conversation_address,
    )

run_accomplice async

run_accomplice(
    ctx: RunContext[None],
    *,
    child: Conversation,
    run_name: str,
    prompt: str,
    model: Model | str | None,
    effort: str | None,
) -> str

Source code in octomate/capabilities/gateway.py

async def run_accomplice(
    self,
    ctx: RunContext[None],
    *,
    child: Conversation,
    run_name: str,
    prompt: str,
    model: Model | str | None,
    effort: str | None,
) -> str:
    agents, conversations, thread_id, conversation_address = self.commission_deps()
    if ctx.run_id is None:
        raise RuntimeError("a commission needs the parent run id for the run tree")
    agent = agents[child.agent_tentacle_id]
    try:
        # The subagent contract (non-interactive, no capabilities,
        # addressed at the pre-ensured child conversation) lives on
        # `subagent_run`; the accomplice framing is this spawner's and is
        # passed here. An accomplice that defers anyway ends its run with
        # DeferredToolRequests, surfaced loudly below instead of parking a
        # batch nothing resumes.
        result = await asyncio.wait_for(
            agent.subagent_run(
                prompt,
                conversation_address=conversation_address,
                thread_id=thread_id,
                conversation_id=child.id,
                run_name=run_name,
                model=model,
                effort=effort,
                instructions=ACCOMPLICE_INSTRUCTION,
            ),
            timeout=self.commission_timeout,
        )
    except TimeoutError:
        raise ModelRetry(
            f"The accomplice {child.subagent_id!r} exceeded "
            f"{self.commission_timeout:.0f}s and was stopped. What it "
            f"recorded is kept — `{GatewayTool.WHISPER}` to continue "
            "it, or break the work into smaller briefs."
        ) from None
    # The runner recorded its turn without knowing its place;
    # the spawner stamps the run tree after the fact.
    await conversations.link_parent_run(
        result.run_id,
        parent_run_id=ctx.run_id,
        parent_tool_call_id=ctx.tool_call_id,
    )
    output = result.output
    if isinstance(output, DeferredToolRequests):
        raise ModelRetry(
            f"The accomplice {child.subagent_id!r} tried to ask the "
            "user or defer — an accomplice has no user. Write a "
            "self-contained brief that needs no clarification, then "
            "commission another."
        )
    if isinstance(output, str):
        return output
    if isinstance(output, Iterable):
        return "\n\n".join(str(part) for part in output)
    return str(output)

inspect async

inspect(
    ctx: RunContext[None],
    reveal: InspectFacet,
    channel: str | None = None,
    inside: str | None = None,
) -> (
    list[AgentRoute]
    | list[ChannelAddress]
    | list[ProjectSummary]
)

Reveal one facet of what this conversation can reach.

Parameters:

Name	Type	Description	Default
reveal	InspectFacet	routes — agent/model choices on the current or selected channel. Summon and commission use the current channel's routes. destinations — optional address suggestions, validated independently when used; known addresses need not appear in this list. With channel, one level of that channel's addresses instead. projects — the projects this deployment can work on, for teleport.	required
channel	str | None	A connected channel ID from an address. For routes, omit it to inspect the current conversation's routes. For destinations, name it to browse that channel as the requesting user: an address whose metadata.inside is set is a place to open, one without it can host a new thread unless metadata.barred says why not.	None
inside	str | None	For destinations with channel, the metadata.inside of a listed address, to list what that place holds. Omit it for the channel's top level.	None

Source code in octomate/capabilities/gateway.py

async def inspect(
    self,
    ctx: RunContext[None],
    reveal: InspectFacet,
    channel: str | None = None,
    inside: str | None = None,
) -> list[AgentRoute] | list[ChannelAddress] | list[ProjectSummary]:
    """Reveal one facet of what this conversation can reach.

    Args:
        reveal: `routes` — agent/model choices on the current or selected channel.
            Summon and commission use the current channel's routes.
            `destinations` — optional address suggestions, validated independently
            when used; known addresses need not appear in this list. With
            `channel`, one level of that channel's addresses instead.
            `projects` — the projects this deployment can work on, for `teleport`.
        channel: A connected channel ID from an address. For `routes`, omit it
            to inspect the current conversation's routes. For `destinations`,
            name it to browse that channel as the requesting user: an address
            whose `metadata.inside` is set is a place to open, one without it
            can host a new thread unless `metadata.barred` says why not.
        inside: For `destinations` with `channel`, the `metadata.inside` of a
            listed address, to list what that place holds. Omit it for the
            channel's top level.
    """
    try:
        return await self.session.inspect(reveal, channel, inside)
    except GatewayRefusal as refusal:
        raise ModelRetry(str(refusal)) from refusal

summon async

summon(
    ctx: RunContext[None],
    agent_id: str,
    model: str,
    hint: str,
    reason: str,
    brief: Annotated[str, Field(max_length=8000)],
    effort: str | None = None,
) -> str

Hand this conversation to another Octomate agent, who takes it over where it is.

Parameters:

Name	Type	Description	Default
agent_id	str	The target agent, copied exactly from an inspect route (reveal="routes").	required
model	str	That route's model, copied exactly.	required
hint	str	A short, user-facing note announcing the handoff, recorded with it.	required
reason	str	One line on why this agent fits — recorded with the handoff, not shown to the user as the reply.	required
brief	Annotated[str, Field(max_length=8000)]	The self-contained brief the other agent starts from. It becomes their opening prompt and they cannot see this conversation, so give the goal, the relevant context and decisions, what's been tried, and what a finished result looks like. If anything you carry says how to hand off — a handoff skill, a plugin, your own harness's instructions — follow it; this is the floor. Reference rather than repeat: the receiver reads every thread this person has spoken in, so cite a message by its #msg:<id> handle and say what to search for instead of pasting it. A brief over the size budget is refused, never trimmed.	required
effort	str | None	How hard the agent should think, from the effort levels the route's claim offers. Set it only when the user explicitly asked for a level; omitted, the agent's own default applies.	None

Source code in octomate/capabilities/gateway.py

async def summon(
    self,
    ctx: RunContext[None],
    agent_id: str,
    model: str,
    hint: str,
    reason: str,
    brief: Annotated[str, Field(max_length=8_000)],
    effort: str | None = None,
) -> str:
    """Hand this conversation to another Octomate agent, who takes it over where
    it is.

    Args:
        agent_id: The target agent, copied exactly from an `inspect` route
            (`reveal="routes"`).
        model: That route's model, copied exactly.
        hint: A short, user-facing note announcing the handoff, recorded with it.
        reason: One line on why this agent fits — recorded with the handoff, not
            shown to the user as the reply.
        brief: The self-contained brief the other agent starts from. It becomes
            their opening prompt and they cannot see this conversation, so give
            the goal, the relevant context and decisions, what's been tried, and
            what a finished result looks like. If anything you carry says how to
            hand off — a handoff skill, a plugin, your own harness's instructions
            — follow it; this is the floor. Reference rather than repeat: the
            receiver reads every thread this person has spoken in, so cite a
            message by its `#msg:<id>` handle and say what to search for instead
            of pasting it. A brief over the size budget is refused, never trimmed.
        effort: How hard the agent should think, from the effort levels the
            route's claim offers. Set it only when the user explicitly asked
            for a level; omitted, the agent's own default applies.
    """
    try:
        return await self.session.summon(
            agent_id=agent_id,
            model=model,
            hint=hint,
            reason=reason,
            brief=brief,
            effort=effort,
        )
    except GatewayRefusal as refusal:
        raise ModelRetry(str(refusal)) from refusal

teleport async

teleport(
    ctx: RunContext[None],
    hint: str,
    destination: ChannelAddress | None = None,
    project: str | None = None,
    ref: str | None = None,
    new_thread: bool = True,
    resume: bool = False,
) -> str

Continue this conversation yourself somewhere else; everything said so far comes with you. This turn ends on it, and your context is intact there.

Parameters:

Name	Type	Description	Default
hint	str	The short, user-facing thread-starter message.	required
destination	ChannelAddress | None	The address at which to create a thread; omit for the current chat. The owning channel validates it independently of discovery and requires the requesting user's linked identity. Shared history stays under its current parent. Teleport requires attached history that the agent can copy.	None
new_thread	bool	Create a thread before continuing. Set false only to bind the current thread to a project without opening another one. For creation, the address must have no channel_thread_id.	True
project	str | None	A project's name, copied exactly from inspect (reveal="projects"), for a conversation about no project: the thread you land in is bound to it, and you resume in its workspace, where work is kept. A conversation about a project carries it wherever it goes, so omit this there.	None
ref	str | None	The branch, tag or commit that workspace starts from; omit it for the project's default branch.	None
resume	bool	Carry on working there at once, as after any tool call. Leave it false when the move is all that was asked: you continue when the user next writes there.	False

Source code in octomate/capabilities/gateway.py

async def teleport(
    self,
    ctx: RunContext[None],
    hint: str,
    destination: ChannelAddress | None = None,
    project: str | None = None,
    ref: str | None = None,
    new_thread: bool = True,
    resume: bool = False,
) -> str:
    """Continue this conversation yourself somewhere else; everything said so
    far comes with you. This turn ends on it, and your context is intact
    there.

    Args:
        hint: The short, user-facing thread-starter message.
        destination: The address at which to create a thread; omit for the
            current chat. The owning channel validates it independently of
            discovery and requires the requesting user's linked identity.
            Shared history stays under its current parent. Teleport requires
            attached history that the agent can copy.
        new_thread: Create a thread before continuing. Set false only to bind
            the current thread to a project without opening another one.
            For creation, the address must have no `channel_thread_id`.
        project: A project's name, copied exactly from `inspect`
            (`reveal="projects"`), for a conversation about no project: the
            thread you land in is bound to it, and you resume in its workspace,
            where work is kept. A conversation about a project carries it
            wherever it goes, so omit this there.
        ref: The branch, tag or commit that workspace starts from; omit it for
            the project's default branch.
        resume: Carry on working there at once, as after any tool call. Leave
            it false when the move is all that was asked: you continue when
            the user next writes there.
    """
    try:
        decision = await self.session.teleport(
            hint=hint,
            destination=destination,
            new_thread=new_thread,
            project=project,
            ref=ref,
            resume=resume,
        )
    except GatewayRefusal as refusal:
        raise ModelRetry(str(refusal)) from refusal
    raise CallDeferred(metadata=decision.metadata())

scheme async

scheme(
    ctx: RunContext[None],
    hint: str,
    brief: Annotated[str, Field(max_length=8000)],
    destination: SchemeTarget = DIRECT_TARGET,
) -> str

Continue this with the user one-to-one, in their direct messages.

Whoever already handles that user's direct messages picks the work up — you do not choose them, and their direct messages keep the owner they had. Use it for work that is that person's alone, or that does not belong in front of the group.

Parameters:

Name	Type	Description	Default
hint	str	The short, user-facing line that opens the conversation over there, where they will read it before anything else. Say that the work is being picked up — not the brief, which is written for whoever answers.	required
brief	Annotated[str, Field(max_length=8000)]	The self-contained brief whoever answers there starts from. They cannot see this conversation, so give the goal, the relevant context and decisions, what's been tried, and what a finished result looks like. If anything you carry says how to hand off — a handoff skill, a plugin, your own harness's instructions — follow it; this is the floor. Reference rather than repeat: the receiver reads every thread this person has spoken in, so cite a message by its #msg:<id> handle and say what to search for instead of pasting it. A brief over the size budget is refused, never trimmed.	required
destination	SchemeTarget	A target object with kind="dm". Omit channel for this channel's DM when moving from a shared source, or name another connected channel ID, including from a private source. The user's linked profile determines the recipient.	DIRECT_TARGET

Source code in octomate/capabilities/gateway.py

async def scheme(
    self,
    ctx: RunContext[None],
    hint: str,
    brief: Annotated[str, Field(max_length=8_000)],
    destination: SchemeTarget = DIRECT_TARGET,
) -> str:
    """Continue this with the user one-to-one, in their direct messages.

    Whoever already handles that user's direct messages picks the work up — you do
    not choose them, and their direct messages keep the owner they had. Use it for
    work that is that person's alone, or that does not belong in front of the group.

    Args:
        hint: The short, user-facing line that opens the conversation over there,
            where they will read it before anything else. Say that the work is
            being picked up — not the brief, which is written for whoever answers.
        brief: The self-contained brief whoever answers there starts from. They
            cannot see this conversation, so give the goal, the relevant context and
            decisions, what's been tried, and what a finished result looks like. If
            anything you carry says how to hand off — a handoff skill, a plugin,
            your own harness's instructions — follow it; this is the floor.
            Reference rather than repeat: the receiver reads every thread this
            person has spoken in, so cite a message by its `#msg:<id>` handle and
            say what to search for instead of pasting it. A brief over the size
            budget is refused, never trimmed.
        destination: A target object with `kind="dm"`. Omit `channel` for this
            channel's DM when moving from a shared source, or name another
            connected channel ID, including from a private source. The user's
            linked profile determines the recipient.
    """
    try:
        return await self.session.scheme(
            hint=hint, brief=brief, destination=destination
        )
    except GatewayRefusal as refusal:
        raise ModelRetry(str(refusal)) from refusal

send async

send(
    ctx: RunContext[None],
    segments: list[MessageSegment],
    destination: SendTarget = HERE_TARGET,
) -> ToolReturn[str]

Deliver these segments immediately, without ending your turn — a progress update, an intermediate result, an image or a file. What you send here is already delivered: do NOT repeat it in your final reply.

Parameters:

Name	Type	Description	Default
segments	list[MessageSegment]	What to deliver.	required
destination	SendTarget	A target object with kind="here" (the default) or kind="dm". A DM target can name a connected channel ID; omit it for this channel. With no current conversation, name a channel explicitly. When choosing from an inspected address, use its channel_tentacle_id as channel; the linked profile supplies the recipient. Say in your reply when you sent it elsewhere.	HERE_TARGET

Source code in octomate/capabilities/gateway.py

async def send(
    self,
    ctx: RunContext[None],
    segments: list[MessageSegment],
    destination: SendTarget = HERE_TARGET,
) -> ToolReturn[str]:
    """Deliver these segments immediately, without ending your turn — a progress
    update, an intermediate result, an image or a file. What you send here is
    already delivered: do NOT repeat it in your final reply.

    Args:
        segments: What to deliver.
        destination: A target object with `kind="here"` (the default) or
            `kind="dm"`. A DM target can name a connected `channel` ID; omit
            it for this channel. With no current conversation, name a channel
            explicitly. When choosing from an inspected address, use its
            `channel_tentacle_id` as `channel`; the linked profile supplies the
            recipient. Say in your reply when you sent it elsewhere.
    """
    try:
        address = await self.session.resolve_send(destination)
    except GatewayRefusal as refusal:
        raise ModelRetry(str(refusal)) from refusal
    return ToolReturn(
        return_value="sent",
        metadata=[MessageSentEvent(segments=segments, destination=address)],
    )

dismiss async

dismiss(ctx: RunContext[None]) -> str

Give this thread's workspace back, now that the work in it is done — merged, delivered, or dropped for good, not paused. It is released when this turn ends, after this turn's work is saved to the project's mirror, so nothing is lost: a later message on this thread forks it afresh from there. Only the disk goes, and it goes now rather than after the idle window the sweep would otherwise wait out.

Source code in octomate/capabilities/gateway.py

async def dismiss(self, ctx: RunContext[None]) -> str:
    """Give this thread's workspace back, now that the work in it is done —
    merged, delivered, or dropped for good, not paused. It is released when
    this turn ends, after this turn's work is saved to the project's mirror,
    so nothing is lost: a later message on this thread forks it afresh from
    there. Only the disk goes, and it goes now rather than after the idle
    window the sweep would otherwise wait out.
    """
    try:
        return await self.session.dismiss()
    except GatewayRefusal as refusal:
        raise ModelRetry(str(refusal)) from refusal

commission async

commission(
    ctx: RunContext[None],
    name: str,
    agent_id: str,
    model: str,
    brief: str,
    effort: str | None = None,
) -> str

Draw another Octomate agent into working a task and get its report back — the user sees none of it.

Parameters:

Name	Type	Description	Default
name	str	Your name for this accomplice — short and mnemonic, e.g. repo-audit. whisper to it later to follow up.	required
agent_id	str	The agent to draw in, copied exactly from an inspect route for this conversation's channel (omit channel).	required
model	str	That route's model, copied exactly.	required
brief	str	The self-contained work order. The accomplice cannot see this conversation and has no user to ask, so give the goal, the relevant context and decisions, and what a finished result looks like.	required
effort	str | None	How hard the accomplice should think, from the levels the route's claim offers. Set it only when the user explicitly asked for a level.	None

Source code in octomate/capabilities/gateway.py

async def commission(
    self,
    ctx: RunContext[None],
    name: str,
    agent_id: str,
    model: str,
    brief: str,
    effort: str | None = None,
) -> str:
    """Draw another Octomate agent into working a task and get its report
    back — the user sees none of it.

    Args:
        name: Your name for this accomplice — short and mnemonic, e.g.
            `repo-audit`. `whisper` to it later to follow up.
        agent_id: The agent to draw in, copied exactly from an
            `inspect` route for this conversation's channel (omit `channel`).
        model: That route's model, copied exactly.
        brief: The self-contained work order. The accomplice cannot see
            this conversation and has no user to ask, so give the
            goal, the relevant context and decisions, and what a
            finished result looks like.
        effort: How hard the accomplice should think, from the levels the
            route's claim offers. Set it only when the user
            explicitly asked for a level.
    """
    agents, conversations, thread_id, _ = self.commission_deps()
    if not name.strip():
        raise ModelRetry("Give the accomplice a short, mnemonic name.")
    if agent_id == self.session.current_agent_id:
        raise ModelRetry(
            f"Cannot commission yourself {self.session.current_agent_id!r}. "
            f'Call `{GatewayTool.INSPECT}` with `reveal="routes"` to choose a valid route.'
        )
    try:
        route = self.session.claimed_route(
            agent_id, model, effort, spell="commission"
        )
    except GatewayRefusal as refusal:
        raise ModelRetry(str(refusal)) from refusal
    run_model = agents[agent_id].models.get(route.model)
    if run_model is None:
        raise ModelRetry(
            f"Agent {agent_id!r} does not serve model {model!r}. "
            f'Call `{GatewayTool.INSPECT}` with `reveal="routes"` and copy a route exactly.'
        )
    # The calling run's own conversation is the parent — the react
    # graph put its id on the RunContext. No id means the gate is
    # mounted outside a live run: raise, never conjure a parent.
    if ctx.conversation_id is None:
        raise RuntimeError("a commission needs the calling run's conversation id")
    parent_id = uuid.UUID(ctx.conversation_id)
    child = await conversations.ensure(
        thread_id,
        agent_tentacle_id=agent_id,
        subagent_id=name,
        parent_conversation_id=parent_id,
    )
    if child.parent_conversation_id != parent_id or child.runs:
        raise ModelRetry(
            f"{name!r} is already at work — `{GatewayTool.WHISPER}` "
            "to it, or pick a new name."
        )
    return await self.run_accomplice(
        ctx,
        child=child,
        run_name="commission",
        prompt=brief,
        model=run_model,
        effort=effort,
    )

whisper async

whisper(
    ctx: RunContext[None], name: str, message: str
) -> str

A quiet word to an accomplice you commissioned. It keeps everything it did — your whisper continues its work, not a fresh start.

Parameters:

Name	Type	Description	Default
name	str	The name you gave the accomplice when you commissioned it.	required
message	str	The follow-up work order — self-contained, like a brief; the accomplice still cannot see this conversation.	required

Source code in octomate/capabilities/gateway.py

async def whisper(self, ctx: RunContext[None], name: str, message: str) -> str:
    """A quiet word to an accomplice you commissioned. It keeps
    everything it did — your whisper continues its work, not a fresh
    start.

    Args:
        name: The name you gave the accomplice when you commissioned it.
        message: The follow-up work order — self-contained, like a
            brief; the accomplice still cannot see this conversation.
    """
    _, conversations, thread_id, _ = self.commission_deps()
    if ctx.conversation_id is None:
        raise RuntimeError("a whisper needs the calling run's conversation id")
    parent_id = uuid.UUID(ctx.conversation_id)
    hands = await conversations.subagents(parent_id)
    stored = next((hand for hand in hands if hand.subagent_id == name), None)
    if stored is None:
        live = ", ".join(sorted(hand.subagent_id for hand in hands))
        raise ModelRetry(
            f"No accomplice named {name!r}. "
            + (
                f"Accomplices at work: {live}."
                if live
                else "No accomplices are at work."
            )
            + f" `{GatewayTool.COMMISSION}` one to start."
        )
    child = await conversations.ensure(
        thread_id,
        agent_tentacle_id=stored.agent_tentacle_id,
        subagent_id=name,
        parent_conversation_id=parent_id,
    )
    return await self.run_accomplice(
        ctx,
        child=child,
        run_name="whisper",
        prompt=message,
        model=None,
        effort=None,
    )

get_instructions

get_instructions() -> str

Source code in octomate/capabilities/gateway.py

def get_instructions(self) -> str:
    instructions = gateway_instructions(lambda name: name) + HANDOFF_INSTRUCTION
    if self.commissioning:
        return instructions + COMMISSION_INSTRUCTION
    return instructions

wrap_run_event_stream async

wrap_run_event_stream(
    ctx: RunContext[None],
    *,
    stream: AsyncIterable[AgentStreamEvent],
) -> AsyncIterable[AgentStreamEvent]

Put each send's stashed MessageSentEvent on the run stream, where the consumer rendering this run delivers it, and each accomplice's start and finish around its call.

Source code in octomate/capabilities/gateway.py

async def wrap_run_event_stream(
    self,
    ctx: RunContext[None],
    *,
    stream: AsyncIterable[AgentStreamEvent],
) -> AsyncIterable[AgentStreamEvent]:
    """Put each `send`'s stashed `MessageSentEvent` on the run stream,
    where the consumer rendering this run delivers it, and each accomplice's
    start and finish around its call."""
    async for event in Accomplices().watch(stream):
        yield event
        if (
            isinstance(event, FunctionToolResultEvent)
            and isinstance(event.part, ToolReturnPart)
            and event.part.tool_name == GatewayTool.SEND
            and isinstance(event.part.metadata, list)
        ):
            for sent_event in event.part.metadata:
                # One dynamic-boundary cast: pydantic-ai types the stream as
                # AgentStreamEvent, consumers match the concrete octomate event.
                yield cast(AgentStreamEvent, sent_event)

get_toolset

get_toolset() -> AbstractToolset[None] | None

Source code in octomate/capabilities/gateway.py

def get_toolset(self) -> AbstractToolset[None] | None:
    return self.toolset

gateway_instructions

gateway_instructions(
    tool_name: Callable[[GatewayTool], str],
) -> str

The gateway's routing instruction, each spell rendered by the caller's own tool naming — the identity for Inkling, gateway_… on the served server — so every agent reads one contract under the names its runtime lists the tools by.

Source code in octomate/capabilities/gateway.py

def gateway_instructions(tool_name: Callable[[GatewayTool], str]) -> str:
    """The gateway's routing instruction, each spell rendered by the caller's own tool
    naming — the identity for Inkling, `gateway_…` on the served server — so every
    agent reads one contract under the names its runtime lists the tools by."""
    names = {
        "inspect": tool_name(GatewayTool.INSPECT),
        "summon": tool_name(GatewayTool.SUMMON),
        "teleport": tool_name(GatewayTool.TELEPORT),
        "scheme": tool_name(GatewayTool.SCHEME),
        "send": tool_name(GatewayTool.SEND),
        "dismiss": tool_name(GatewayTool.DISMISS),
    }
    return GATEWAY_INSTRUCTION_TEMPLATE.format(
        **names
    ) + SEND_INSTRUCTION_TEMPLATE.format(**names)

octomate.capabilities.ask

Ask capability: the deferred ask-the-user tool.

A capability rather than a loose toolset entry so the caller decides per run whether a user exists to ask at all — an interactive run mounts it, an accomplice run simply never sees the tool.

ASK_QUESTIONS_TOOL_NAME module-attribute

ASK_QUESTIONS_TOOL_NAME = 'ask_questions'

ASK_DEFERR_KIND module-attribute

ASK_DEFERR_KIND = 'questions'

AskCapability dataclass

Bases: AbstractCapability[None]

Offers ask_questions — the deferred ask-the-user batch.

Source code in octomate/capabilities/ask.py

@dataclass
class AskCapability(AbstractCapability[None]):
    """Offers `ask_questions` — the deferred ask-the-user batch."""

    toolset: FunctionToolset[None] | None = field(default=None, init=False, repr=False)

    def __post_init__(self) -> None:
        toolset: FunctionToolset[None] = FunctionToolset(id="ask")

        @toolset.tool(name=ASK_QUESTIONS_TOOL_NAME)
        async def ask_questions(
            ctx: RunContext[None],
            questions: list[QuestionRequest],
        ) -> list[str]:
            """Ask the user several questions and wait for all answers as one batch."""
            if not questions:
                raise ValueError("ask_questions requires at least one question")
            raise CallDeferred(metadata={"kind": ASK_DEFERR_KIND})

        self.toolset = toolset

    def get_toolset(self) -> AbstractToolset[None] | None:
        return self.toolset

toolset class-attribute instance-attribute

toolset: FunctionToolset[None] | None = field(
    default=None, init=False, repr=False
)

get_toolset

get_toolset() -> AbstractToolset[None] | None

Source code in octomate/capabilities/ask.py

def get_toolset(self) -> AbstractToolset[None] | None:
    return self.toolset

octomate.capabilities.history

History capability: the thread ledger of the person a run answers.

Thread history is what people, bots and agents visibly said — ThreadMessage rows, including the ones that did not wake this agent. The model ledger is not offered here: it exists for audit and for rebuilding a conversation's context, not for a model to read.

The scope is the person, not the thread: a run reads every thread the human it is answering has spoken in, on any account the registry links to them — this thread, their direct messages elsewhere, the chat a handoff came from. So the capability is user-scoped, bound per run by for_profile, and no tool takes a conversation id.

The tools are methods with docstrings, as the gateway's spells are, so a runtime that reaches them some other way than pydantic-ai projects the same contracts.

HISTORY_TOOLSET_ID module-attribute

HISTORY_TOOLSET_ID = 'history'

HISTORY_INSTRUCTION_TEMPLATE module-attribute

HISTORY_INSTRUCTION_TEMPLATE = "## Searching history\n\nThread history is what people, bots, and agents visibly said — in this thread, and\nin every other thread the person you are answering has spoken in, on any of their\nlinked accounts. Messages that did not wake you are there too.\n\n- `{search_thread_history}` searches all of it for English terms, requiring every\n  term and ranking the best lexical matches first. It accepts an optional actor\n  kind. Prefer it when the user refers to what was said, here or elsewhere.\n- `{read_thread_history_before}` and `{read_thread_history_after}` page a thread\n  around a search hit, or around a `#msg:<id>` handle a brief cited.\n"

HISTORY_TOOLS module-attribute

HISTORY_TOOLS: tuple[str, ...] = (
    "search_thread_history",
    "read_thread_history_before",
    "read_thread_history_after",
)

HISTORY_INSTRUCTIONS module-attribute

HISTORY_INSTRUCTIONS = history_instructions(
    lambda name: name
)

HistoryCapability dataclass

Bases: AbstractCapability[Any]

Thread-history tools over the history of the person a run answers.

Mounted user-scoped: the tentacle asks for_profile for the copy bound to that run's user and mounts that, so no tool has to ask who is asking.

Source code in octomate/capabilities/history.py

@dataclass
class HistoryCapability(AbstractCapability[Any]):
    """Thread-history tools over the history of the person a run answers.

    Mounted user-scoped: the tentacle asks `for_profile` for the copy bound to
    that run's user and mounts that, so no tool has to ask who is asking.
    """

    # No default: the capability must read the host's ledger manager, never a
    # private one with its own identity registry.
    thread_manager: ThreadManager
    # The person whose history the tools read. None on the mounted template,
    # which serves no run itself.
    profile: UserProfile | None = None
    toolset: FunctionToolset[Any] | None = field(default=None, init=False, repr=False)

    def __post_init__(self) -> None:
        toolset: FunctionToolset[Any] = FunctionToolset(id=HISTORY_TOOLSET_ID)
        # No tool takes the run context: the person is bound in, not read off it.
        toolset.add_function(self.search_thread_history, takes_ctx=False)
        toolset.add_function(self.read_thread_history_before, takes_ctx=False)
        toolset.add_function(self.read_thread_history_after, takes_ctx=False)
        self.toolset = toolset

    async def for_profile(self, profile: UserProfile) -> HistoryCapability:
        """This capability bound to one person. Everyone gets one: a visitor's
        history is the threads their one account has spoken in."""
        return replace(self, profile=profile)

    @property
    def reader(self) -> UserProfile:
        if self.profile is None:
            raise RuntimeError(
                "history is read as a run's user: mount the copy `for_profile` gives"
            )
        return self.profile

    async def anchor(self, handle: str) -> ThreadMessage:
        """The message a paging tool is anchored on, with a refusal spoken as a
        retry: a handle outside this person's history is not theirs to page."""
        try:
            return await self.thread_manager.chat_message(self.reader, handle)
        except ValueError as refusal:
            raise ModelRetry(str(refusal)) from refusal

    async def search_thread_history(
        self,
        query: str,
        actor_kind: ChannelActorKind | None = None,
        limit: int = 10,
    ) -> list[ThreadMessage]:
        """Find visible thread messages matching every English term in `query`,
        with related word forms matched by stemming, across every thread the person
        you are answering has spoken in — this one, their direct messages, other
        chats, on any of their linked accounts. Optionally restrict to an actor kind
        such as "human", "agent", "bot", or "system". Best lexical match first,
        newest first when scores tie. Use a hit's id to page its local context."""
        try:
            return await self.thread_manager.search_chat_messages(
                self.reader, query, actor_kind=actor_kind, limit=limit
            )
        except ValueError as refusal:
            raise ModelRetry(str(refusal)) from refusal

    async def read_thread_history_before(
        self, message_id: str, limit: int = 10
    ) -> list[ThreadMessage]:
        """The visible thread messages immediately preceding `message_id` in its
        thread, oldest first. `message_id` is a row id or a `#msg:<id>` handle, as
        a search hit or a brief shows it; a message outside this person's history
        is refused."""
        anchor = await self.anchor(message_id)
        return await self.thread_manager.chat_messages_before(
            anchor.thread_id, anchor.id, limit=limit
        )

    async def read_thread_history_after(
        self, message_id: str, limit: int = 10
    ) -> list[ThreadMessage]:
        """The visible thread messages immediately following `message_id` in its
        thread, oldest first. `message_id` is a row id or a `#msg:<id>` handle, as
        a search hit or a brief shows it; a message outside this person's history
        is refused."""
        anchor = await self.anchor(message_id)
        return await self.thread_manager.chat_messages_after(
            anchor.thread_id, anchor.id, limit=limit
        )

    def get_toolset(self) -> FunctionToolset[Any] | None:
        return self.toolset

    def get_instructions(self) -> AgentInstructions[Any] | None:
        return HISTORY_INSTRUCTIONS

thread_manager instance-attribute

thread_manager: ThreadManager

profile class-attribute instance-attribute

profile: UserProfile | None = None

toolset class-attribute instance-attribute

toolset: FunctionToolset[Any] | None = field(
    default=None, init=False, repr=False
)

reader property

reader: UserProfile

for_profile async

for_profile(profile: UserProfile) -> HistoryCapability

This capability bound to one person. Everyone gets one: a visitor's history is the threads their one account has spoken in.

Source code in octomate/capabilities/history.py

async def for_profile(self, profile: UserProfile) -> HistoryCapability:
    """This capability bound to one person. Everyone gets one: a visitor's
    history is the threads their one account has spoken in."""
    return replace(self, profile=profile)

anchor async

anchor(handle: str) -> ThreadMessage

The message a paging tool is anchored on, with a refusal spoken as a retry: a handle outside this person's history is not theirs to page.

Source code in octomate/capabilities/history.py

async def anchor(self, handle: str) -> ThreadMessage:
    """The message a paging tool is anchored on, with a refusal spoken as a
    retry: a handle outside this person's history is not theirs to page."""
    try:
        return await self.thread_manager.chat_message(self.reader, handle)
    except ValueError as refusal:
        raise ModelRetry(str(refusal)) from refusal

search_thread_history async

search_thread_history(
    query: str,
    actor_kind: ChannelActorKind | None = None,
    limit: int = 10,
) -> list[ThreadMessage]

Find visible thread messages matching every English term in query, with related word forms matched by stemming, across every thread the person you are answering has spoken in — this one, their direct messages, other chats, on any of their linked accounts. Optionally restrict to an actor kind such as "human", "agent", "bot", or "system". Best lexical match first, newest first when scores tie. Use a hit's id to page its local context.

Source code in octomate/capabilities/history.py

async def search_thread_history(
    self,
    query: str,
    actor_kind: ChannelActorKind | None = None,
    limit: int = 10,
) -> list[ThreadMessage]:
    """Find visible thread messages matching every English term in `query`,
    with related word forms matched by stemming, across every thread the person
    you are answering has spoken in — this one, their direct messages, other
    chats, on any of their linked accounts. Optionally restrict to an actor kind
    such as "human", "agent", "bot", or "system". Best lexical match first,
    newest first when scores tie. Use a hit's id to page its local context."""
    try:
        return await self.thread_manager.search_chat_messages(
            self.reader, query, actor_kind=actor_kind, limit=limit
        )
    except ValueError as refusal:
        raise ModelRetry(str(refusal)) from refusal

read_thread_history_before async

read_thread_history_before(
    message_id: str, limit: int = 10
) -> list[ThreadMessage]

The visible thread messages immediately preceding message_id in its thread, oldest first. message_id is a row id or a #msg:<id> handle, as a search hit or a brief shows it; a message outside this person's history is refused.

Source code in octomate/capabilities/history.py

async def read_thread_history_before(
    self, message_id: str, limit: int = 10
) -> list[ThreadMessage]:
    """The visible thread messages immediately preceding `message_id` in its
    thread, oldest first. `message_id` is a row id or a `#msg:<id>` handle, as
    a search hit or a brief shows it; a message outside this person's history
    is refused."""
    anchor = await self.anchor(message_id)
    return await self.thread_manager.chat_messages_before(
        anchor.thread_id, anchor.id, limit=limit
    )

read_thread_history_after async

read_thread_history_after(
    message_id: str, limit: int = 10
) -> list[ThreadMessage]

The visible thread messages immediately following message_id in its thread, oldest first. message_id is a row id or a #msg:<id> handle, as a search hit or a brief shows it; a message outside this person's history is refused.

Source code in octomate/capabilities/history.py

async def read_thread_history_after(
    self, message_id: str, limit: int = 10
) -> list[ThreadMessage]:
    """The visible thread messages immediately following `message_id` in its
    thread, oldest first. `message_id` is a row id or a `#msg:<id>` handle, as
    a search hit or a brief shows it; a message outside this person's history
    is refused."""
    anchor = await self.anchor(message_id)
    return await self.thread_manager.chat_messages_after(
        anchor.thread_id, anchor.id, limit=limit
    )

get_toolset

get_toolset() -> FunctionToolset[Any] | None

Source code in octomate/capabilities/history.py

def get_toolset(self) -> FunctionToolset[Any] | None:
    return self.toolset

get_instructions

get_instructions() -> AgentInstructions[Any] | None

Source code in octomate/capabilities/history.py

def get_instructions(self) -> AgentInstructions[Any] | None:
    return HISTORY_INSTRUCTIONS

history_instructions

history_instructions(
    tool_name: Callable[[str], str],
) -> str

The history instruction, each tool rendered by the caller's own naming.

Source code in octomate/capabilities/history.py

def history_instructions(tool_name: Callable[[str], str]) -> str:
    """The history instruction, each tool rendered by the caller's own naming."""
    return HISTORY_INSTRUCTION_TEMPLATE.format(
        **{name: tool_name(name) for name in HISTORY_TOOLS}
    )

conversation_id

conversation_id(ctx: object) -> UUID7

Source code in octomate/capabilities/history.py

def conversation_id(ctx: object) -> UUID7:
    value = getattr(ctx, "conversation_id", None)
    if isinstance(value, uuid.UUID):
        return value
    if isinstance(value, str):
        return uuid.UUID(value)
    if value is None:
        raise ValueError("history tools require a conversation_id on the run")
    raise TypeError("history tools require a UUID conversation_id")

thread_id async

thread_id(
    ctx: object, conversation_manager: ConversationManager
) -> UUID7

Source code in octomate/capabilities/history.py

async def thread_id(ctx: object, conversation_manager: ConversationManager) -> UUID7:
    value = await conversation_manager.thread_id(conversation_id(ctx))
    if value is not None:
        return value
    raise ValueError("thread history tools require a thread-backed conversation")

octomate.capabilities.todos

Todo capability: a persisted, conversation-scoped task list the agent manages.

Models the pydantic-ai-todo toolset (read/write/add/update/remove + subtasks & dependencies) but persists via Arcanus (TodoManager) and emits granular todo events onto the octomate event stream instead of a private callback bus. The capability holds the manager; each tool reads ctx.conversation_id for scoping, so the agent's deps type is untouched. Tools stash the events they produce on ToolReturn.metadata; wrap_run_event_stream forwards them onto the stream.

TODO_TOOL_NAMES module-attribute

TODO_TOOL_NAMES = frozenset(
    {
        "read_todos",
        "write_todos",
        "add_todo",
        "update_todo_status",
        "remove_todo",
        "add_subtask",
        "set_dependency",
        "get_available_tasks",
    }
)

TODO_INSTRUCTION module-attribute

TODO_INSTRUCTION = '## Task management\n\nYou have todo tools to plan and track multi-step work:\n- `read_todos` — view the current tasks with their ids and statuses.\n- `write_todos` — replace the whole list (use to lay out or restructure a plan).\n- `add_todo` — append one task without disturbing the rest.\n- `update_todo_status` — move a task between pending / in_progress / completed.\n- `remove_todo` — delete a task by id.\n- `add_subtask` / `set_dependency` / `get_available_tasks` — break tasks down,\n  order them, and find what is unblocked.\n\nUse them for non-trivial work (3+ steps). Keep exactly one task `in_progress`,\nmark tasks `completed` immediately when done, and pass each task an imperative\n`content` ("Run tests") plus a present-continuous `active_form` ("Running tests").\n'

TodoItem pydantic-model

Bases: BaseModel

One todo in a write_todos call (the model rewrites the whole list).

Fields:

• content (str)
• status (TodoStatus)
• active_form (str)
• id (str | None)
• parent_id (str | None)
• depends_on (list[str])

Source code in octomate/capabilities/todos.py

class TodoItem(BaseModel):
    """One todo in a `write_todos` call (the model rewrites the whole list)."""

    content: str = Field(
        description="Imperative task title, e.g. 'Fix the login bug'.",
    )
    status: TodoStatus = Field(
        default="pending",
        description="pending, in_progress, completed, or blocked.",
    )
    active_form: str = Field(
        description="Present-continuous label, e.g. 'Fixing the login bug'.",
    )
    id: str | None = Field(
        default=None,
        description="Existing todo id to keep; omit to create a new one.",
    )
    parent_id: str | None = Field(
        default=None,
        description="Parent todo id when this is a subtask.",
    )
    depends_on: list[str] = Field(
        default_factory=list,
        description="Ids of todos that must complete before this one.",
    )

content pydantic-field

content: str

Imperative task title, e.g. 'Fix the login bug'.

status pydantic-field

status: TodoStatus = 'pending'

pending, in_progress, completed, or blocked.

active_form pydantic-field

active_form: str

Present-continuous label, e.g. 'Fixing the login bug'.

id pydantic-field

id: str | None = None

Existing todo id to keep; omit to create a new one.

parent_id pydantic-field

parent_id: str | None = None

Parent todo id when this is a subtask.

depends_on pydantic-field

depends_on: list[str]

Ids of todos that must complete before this one.

TodoCapability dataclass

Bases: AbstractCapability[Any]

Adds the persisted todo toolset + its instruction, and emits granular todo events onto the stream as the model mutates the list.

Source code in octomate/capabilities/todos.py

@dataclass
class TodoCapability(AbstractCapability[Any]):
    """Adds the persisted todo toolset + its instruction, and emits granular todo
    events onto the stream as the model mutates the list."""

    manager: TodoManager = field(default_factory=TodoManager)
    enable_subtasks: bool = True
    toolset: FunctionToolset[Any] | None = field(default=None, init=False, repr=False)

    def __post_init__(self) -> None:
        self.toolset = build_todo_toolset(
            self.manager, enable_subtasks=self.enable_subtasks
        )

    def get_toolset(self) -> AbstractToolset[Any] | None:
        return self.toolset

    def get_instructions(self) -> AgentInstructions[Any] | None:
        return TODO_INSTRUCTION

    async def wrap_run_event_stream(
        self,
        ctx: RunContext[Any],
        *,
        stream: AsyncIterable[AgentStreamEvent],
    ) -> AsyncIterable[AgentStreamEvent]:
        async for event in stream:
            yield event
            if (
                isinstance(event, FunctionToolResultEvent)
                and isinstance(event.part, ToolReturnPart)
                and event.part.tool_name in TODO_TOOL_NAMES
                and isinstance(event.part.metadata, list)
            ):
                for todo_event in event.part.metadata:
                    # The tool stashed our display events (not AgentStreamEvents) in
                    # metadata; inject them on the stream. One dynamic-boundary cast —
                    # pydantic-ai types the stream as AgentStreamEvent, consumers
                    # match the concrete octomate event type.
                    yield cast(AgentStreamEvent, todo_event)

manager class-attribute instance-attribute

manager: TodoManager = field(default_factory=TodoManager)

enable_subtasks class-attribute instance-attribute

enable_subtasks: bool = True

toolset class-attribute instance-attribute

toolset: FunctionToolset[Any] | None = field(
    default=None, init=False, repr=False
)

get_toolset

get_toolset() -> AbstractToolset[Any] | None

Source code in octomate/capabilities/todos.py

def get_toolset(self) -> AbstractToolset[Any] | None:
    return self.toolset

get_instructions

get_instructions() -> AgentInstructions[Any] | None

Source code in octomate/capabilities/todos.py

def get_instructions(self) -> AgentInstructions[Any] | None:
    return TODO_INSTRUCTION

wrap_run_event_stream async

wrap_run_event_stream(
    ctx: RunContext[Any],
    *,
    stream: AsyncIterable[AgentStreamEvent],
) -> AsyncIterable[AgentStreamEvent]

Source code in octomate/capabilities/todos.py

async def wrap_run_event_stream(
    self,
    ctx: RunContext[Any],
    *,
    stream: AsyncIterable[AgentStreamEvent],
) -> AsyncIterable[AgentStreamEvent]:
    async for event in stream:
        yield event
        if (
            isinstance(event, FunctionToolResultEvent)
            and isinstance(event.part, ToolReturnPart)
            and event.part.tool_name in TODO_TOOL_NAMES
            and isinstance(event.part.metadata, list)
        ):
            for todo_event in event.part.metadata:
                # The tool stashed our display events (not AgentStreamEvents) in
                # metadata; inject them on the stream. One dynamic-boundary cast —
                # pydantic-ai types the stream as AgentStreamEvent, consumers
                # match the concrete octomate event type.
                yield cast(AgentStreamEvent, todo_event)

by_ref

by_ref(todos: list[Todo]) -> dict[str, Todo]

Source code in octomate/capabilities/todos.py

def by_ref(todos: list[Todo]) -> dict[str, Todo]:
    return {todo.ref: todo for todo in todos}

is_blocked

is_blocked(
    todo: Todo, todos_by_ref: dict[str, Todo]
) -> bool

Source code in octomate/capabilities/todos.py

def is_blocked(todo: Todo, todos_by_ref: dict[str, Todo]) -> bool:
    return any(
        (dep := todos_by_ref.get(ref)) is not None and dep.status != "completed"
        for ref in todo.depends_on
    )

has_cycle

has_cycle(
    todos: list[Todo], todo_ref: str, depends_on_ref: str
) -> bool

Would making todo_ref depend on depends_on_ref create a cycle?

Source code in octomate/capabilities/todos.py

def has_cycle(todos: list[Todo], todo_ref: str, depends_on_ref: str) -> bool:
    """Would making `todo_ref` depend on `depends_on_ref` create a cycle?"""
    todos_by_ref = by_ref(todos)
    visited: set[str] = set()

    def visit(current: str) -> bool:
        if current == todo_ref:
            return True
        if current in visited:
            return False
        visited.add(current)
        dep = todos_by_ref.get(current)
        if dep is None:
            return False
        return any(visit(ref) for ref in dep.depends_on)

    return visit(depends_on_ref)

available_tasks

available_tasks(todos: list[Todo]) -> list[Todo]

Source code in octomate/capabilities/todos.py

def available_tasks(todos: list[Todo]) -> list[Todo]:
    todos_by_ref = by_ref(todos)
    return [
        todo
        for todo in todos
        if todo.status not in ("completed", "blocked")
        and not is_blocked(todo, todos_by_ref)
    ]

render_todos

render_todos(
    todos: list[Todo], *, hierarchical: bool
) -> str

Source code in octomate/capabilities/todos.py

def render_todos(todos: list[Todo], *, hierarchical: bool) -> str:
    if hierarchical:
        children: dict[str | None, list[Todo]] = {}
        for todo in todos:
            children.setdefault(todo.parent_ref, []).append(todo)
        lines = ["Current todos (tree):"]
        counter = [0]

        def walk(parent: str | None, depth: int) -> None:
            for todo in children.get(parent, []):
                counter[0] += 1
                indent = "  " * depth
                marker = STATUS_MARKERS.get(todo.status, "[ ]")
                lines.append(
                    f"{indent}{counter[0]}. {marker} [{todo.ref}] {todo.content}"
                )
                if todo.depends_on:
                    lines.append(f"{indent}   depends on {', '.join(todo.depends_on)}")
                walk(todo.ref, depth + 1)

        walk(None, 0)
    else:
        lines = ["Current todos:"]
        for index, todo in enumerate(todos, 1):
            marker = STATUS_MARKERS.get(todo.status, "[ ]")
            lines.append(f"{index}. {marker} [{todo.ref}] {todo.content}")
            if todo.parent_ref:
                lines.append(f"   (subtask of {todo.parent_ref})")
            if todo.depends_on:
                lines.append(f"   (depends on {', '.join(todo.depends_on)})")

    counts = Counter(todo.status for todo in todos)
    summary = (
        f"\nSummary: {counts['completed']} completed, {counts['blocked']} blocked, "
        f"{counts['in_progress']} in progress, {counts['pending']} pending"
    )
    return "\n".join(lines) + summary

update_events

update_events(
    current: Todo, previous: Todo
) -> list[TodoEvent]

Source code in octomate/capabilities/todos.py

def update_events(current: Todo, previous: Todo) -> list[TodoEvent]:
    events: list[TodoEvent] = [TodoUpdatedEvent(todo=current, previous=previous)]
    if current.status != previous.status:
        events.append(TodoStatusChangedEvent(todo=current, previous=previous))
        if current.status == "completed":
            events.append(TodoCompletedEvent(todo=current))
    return events

build_todo_toolset

build_todo_toolset(
    manager: TodoManager, *, enable_subtasks: bool
) -> FunctionToolset[Any]

Source code in octomate/capabilities/todos.py

def build_todo_toolset(
    manager: TodoManager, *, enable_subtasks: bool
) -> FunctionToolset[Any]:
    toolset: FunctionToolset[Any] = FunctionToolset(id="todo")

    def conversation_id(ctx: RunContext[Any]) -> UUID7:
        if ctx.conversation_id is None:
            raise ValueError("todo tools require a conversation_id on the run")
        return uuid.UUID(ctx.conversation_id)

    @toolset.tool
    async def read_todos(ctx: RunContext[Any], hierarchical: bool = False) -> str:
        """List the current todos with their ids and statuses."""
        todos = await manager.list_todos(conversation_id(ctx))
        if not todos:
            return "No todos yet. Use write_todos or add_todo to create some."
        return render_todos(todos, hierarchical=hierarchical)

    @toolset.tool
    async def write_todos(
        ctx: RunContext[Any], todos: list[TodoItem]
    ) -> ToolReturn[str]:
        """Replace the whole todo list with `todos`."""
        items: list[TodoWrite] = [
            TodoWrite(
                content=item.content,
                status=item.status,
                active_form=item.active_form,
                ref=item.id,
                parent_ref=item.parent_id,
                depends_on=item.depends_on,
            )
            for item in todos
        ]
        deleted, created = await manager.write_todos(conversation_id(ctx), items)
        events: list[TodoEvent] = [TodoDeletedEvent(todo=todo) for todo in deleted]
        events += [TodoCreatedEvent(todo=todo) for todo in created]
        return ToolReturn(return_value=f"Wrote {len(created)} todos.", metadata=events)

    @toolset.tool
    async def add_todo(
        ctx: RunContext[Any], content: str, active_form: str
    ) -> ToolReturn[str]:
        """Append a single new todo (status pending)."""
        todo = await manager.add_todo(
            conversation_id(ctx), content=content, active_form=active_form
        )
        return ToolReturn(
            return_value=f"Added '{content}' (id: {todo.ref}).",
            metadata=[TodoCreatedEvent(todo=todo)],
        )

    @toolset.tool
    async def update_todo_status(
        ctx: RunContext[Any], todo_id: str, status: TodoStatus
    ) -> str | ToolReturn[str]:
        """Change a todo's status by id."""
        conv = conversation_id(ctx)
        previous = await manager.get_todo(conv, todo_id)
        if previous is None:
            return f"No todo with id '{todo_id}'."
        if enable_subtasks and status == "in_progress":
            if is_blocked(previous, by_ref(await manager.list_todos(conv))):
                return (
                    f"Cannot start '{previous.content}' — it has incomplete "
                    "dependencies."
                )
        updated = await manager.update_todo(conv, todo_id, status=status)
        if updated is None:
            return f"No todo with id '{todo_id}'."
        return ToolReturn(
            return_value=f"Updated '{updated.content}' → {status}.",
            metadata=update_events(updated, previous),
        )

    @toolset.tool
    async def remove_todo(ctx: RunContext[Any], todo_id: str) -> str | ToolReturn[str]:
        """Delete a todo by id."""
        removed = await manager.remove_todo(conversation_id(ctx), todo_id)
        if removed is None:
            return f"No todo with id '{todo_id}'."
        return ToolReturn(
            return_value=f"Removed '{removed.content}'.",
            metadata=[TodoDeletedEvent(todo=removed)],
        )

    if enable_subtasks:

        @toolset.tool
        async def add_subtask(
            ctx: RunContext[Any], parent_id: str, content: str, active_form: str
        ) -> str | ToolReturn[str]:
            """Add a subtask under an existing todo."""
            conv = conversation_id(ctx)
            parent = await manager.get_todo(conv, parent_id)
            if parent is None:
                return f"No parent todo with id '{parent_id}'."
            todo = await manager.add_todo(
                conv, content=content, active_form=active_form, parent_ref=parent_id
            )
            return ToolReturn(
                return_value=f"Added subtask '{content}' (id: {todo.ref}, parent: {parent_id}).",
                metadata=[TodoCreatedEvent(todo=todo)],
            )

        @toolset.tool
        async def set_dependency(
            ctx: RunContext[Any], todo_id: str, depends_on_id: str
        ) -> str | ToolReturn[str]:
            """Make `todo_id` depend on `depends_on_id` (blocks it until that completes)."""
            conv = conversation_id(ctx)
            todos = await manager.list_todos(conv)
            todos_by_ref = by_ref(todos)
            todo = todos_by_ref.get(todo_id)
            dependency = todos_by_ref.get(depends_on_id)
            if todo is None:
                return f"No todo with id '{todo_id}'."
            if dependency is None:
                return f"No todo with id '{depends_on_id}'."
            if todo_id == depends_on_id:
                return "A todo cannot depend on itself."
            if has_cycle(todos, todo_id, depends_on_id):
                return "Cannot add dependency: it would create a cycle."
            if depends_on_id in todo.depends_on:
                return "That dependency already exists."
            new_status: TodoStatus = todo.status
            if dependency.status != "completed" and todo.status not in (
                "completed",
                "blocked",
            ):
                new_status = "blocked"
            updated = await manager.update_todo(
                conv,
                todo_id,
                depends_on=[*todo.depends_on, depends_on_id],
                status=new_status,
            )
            if updated is None:
                return f"No todo with id '{todo_id}'."
            message = f"'{todo.content}' now depends on '{dependency.content}'."
            if new_status == "blocked" and todo.status != "blocked":
                message += " Task blocked until the dependency completes."
            return ToolReturn(
                return_value=message, metadata=update_events(updated, todo)
            )

        @toolset.tool
        async def get_available_tasks(ctx: RunContext[Any]) -> str:
            """List tasks with no incomplete dependencies (workable now)."""
            available = available_tasks(await manager.list_todos(conversation_id(ctx)))
            if not available:
                return "No available tasks — all are completed or blocked."
            lines = ["Available tasks:"]
            for index, todo in enumerate(available, 1):
                marker = STATUS_MARKERS.get(todo.status, "[ ]")
                lines.append(f"{index}. {marker} [{todo.ref}] {todo.content}")
            return "\n".join(lines)

    return toolset

octomate.capabilities.mcp

Octomate's MCP tentacles as an Inkling capability: the tools the served server lists a turn, mounted in process for one run.

Inkling runs in this process, so nothing goes over the wire — not even in memory: the tentacles' server is called as the object it is, no client and no transport between them. The run's Octomate session is closed over, every tentacle lists and calls as the person that session is for, and the linking pair is there beside them. The spells and the history Inkling has as capabilities of its own, so what it mounts here is the tentacles' server alone, deferred: named by a catalog line, its tools flagged for the model to discover or load before they are on the wire, so a listing that differs per person never touches the prompt prefix.

ANY_ARGUMENTS module-attribute

ANY_ARGUMENTS = SchemaValidator(core_schema.any_schema())

TentaclesToolset

Bases: AbstractToolset[None]

A FastMCP server's tools as a pydantic-ai toolset, the server called in process as the object it is. Named as the server is, and its instructions are the server's own.

Source code in octomate/capabilities/mcp.py

class TentaclesToolset(AbstractToolset[None]):
    """A FastMCP server's tools as a pydantic-ai toolset, the server called in
    process as the object it is. Named as the server is, and its instructions
    are the server's own."""

    def __init__(self, server: FastMCP) -> None:
        self.server = server

    @property
    def id(self) -> str | None:
        return self.server.name

    async def get_instructions(self, ctx: RunContext[None]) -> str | None:
        return self.server.instructions

    async def get_tools(self, ctx: RunContext[None]) -> dict[str, ToolsetTool[None]]:
        return {
            tool.name: ToolsetTool(
                toolset=self,
                tool_def=ToolDefinition(
                    name=tool.name,
                    description=tool.description,
                    parameters_json_schema=tool.parameters,
                ),
                max_retries=ctx.max_retries,
                args_validator=ANY_ARGUMENTS,
            )
            for tool in await self.server.list_tools()
        }

    async def call_tool(
        self,
        name: str,
        tool_args: dict[str, Any],
        ctx: RunContext[None],
        tool: ToolsetTool[None],
    ) -> dict[str, Any] | str:
        try:
            result = await self.server.call_tool(name, tool_args)
        except (ToolError, ValidationError) as refusal:
            # The same corrective sentence every runtime reads, as the retry
            # Inkling corrects from.
            raise ModelRetry(str(refusal)) from refusal
        # What the upstream said, as it said it: text when it spoke in text —
        # FastMCP wraps a bare value as `{"result": …}` beside it, which is not
        # the upstream's — and its structured answer otherwise.
        texts = [
            block.text for block in result.content if isinstance(block, TextContent)
        ]
        if result.is_error:
            raise ModelRetry("\n".join(texts) or "MCP tool call failed")
        if len(texts) == len(result.content):
            return "\n".join(texts)
        if result.structured_content is not None:
            return result.structured_content
        return "\n".join(str(block) for block in result.content)

server instance-attribute

server = server

id property

id: str | None

get_instructions async

get_instructions(ctx: RunContext[None]) -> str | None

Source code in octomate/capabilities/mcp.py

async def get_instructions(self, ctx: RunContext[None]) -> str | None:
    return self.server.instructions

get_tools async

get_tools(
    ctx: RunContext[None],
) -> dict[str, ToolsetTool[None]]

Source code in octomate/capabilities/mcp.py

async def get_tools(self, ctx: RunContext[None]) -> dict[str, ToolsetTool[None]]:
    return {
        tool.name: ToolsetTool(
            toolset=self,
            tool_def=ToolDefinition(
                name=tool.name,
                description=tool.description,
                parameters_json_schema=tool.parameters,
            ),
            max_retries=ctx.max_retries,
            args_validator=ANY_ARGUMENTS,
        )
        for tool in await self.server.list_tools()
    }

call_tool async

call_tool(
    name: str,
    tool_args: dict[str, Any],
    ctx: RunContext[None],
    tool: ToolsetTool[None],
) -> dict[str, Any] | str

Source code in octomate/capabilities/mcp.py

async def call_tool(
    self,
    name: str,
    tool_args: dict[str, Any],
    ctx: RunContext[None],
    tool: ToolsetTool[None],
) -> dict[str, Any] | str:
    try:
        result = await self.server.call_tool(name, tool_args)
    except (ToolError, ValidationError) as refusal:
        # The same corrective sentence every runtime reads, as the retry
        # Inkling corrects from.
        raise ModelRetry(str(refusal)) from refusal
    # What the upstream said, as it said it: text when it spoke in text —
    # FastMCP wraps a bare value as `{"result": …}` beside it, which is not
    # the upstream's — and its structured answer otherwise.
    texts = [
        block.text for block in result.content if isinstance(block, TextContent)
    ]
    if result.is_error:
        raise ModelRetry("\n".join(texts) or "MCP tool call failed")
    if len(texts) == len(result.content):
        return "\n".join(texts)
    if result.structured_content is not None:
        return result.structured_content
    return "\n".join(str(block) for block in result.content)

tentacles_capability

tentacles_capability(
    session: OctomateSession, *, manager: McpManager
) -> Toolset[None]

The capability a run mounts: the tentacles' server built over session, deferred behind a catalog line naming what it holds.

Source code in octomate/capabilities/mcp.py

def tentacles_capability(
    session: OctomateSession,
    *,
    manager: McpManager,
) -> Toolset[None]:
    """The capability a run mounts: the tentacles' server built over `session`,
    deferred behind a catalog line naming what it holds."""

    async def fixed() -> OctomateSession:
        return session

    server = tentacles_mcp(fixed, manager=manager)
    return Toolset(
        TentaclesToolset(server),
        id=server.name,
        description=(
            "The user's installed MCP tools, "
            "acting as the person you are answering, and "
            "the linking of their accounts."
        ),
        defer_loading=True,
    )

octomate.capabilities.tools

Tool-level capabilities: reporting a failed tool call to the model instead of ending the turn.

ToolFailureCapability dataclass

Bases: AbstractCapability[None]

Report a failed tool to the model instead of ending the turn.

Pydantic AI ends a run when a tool raises anything but its own control-flow signals, so one broken call — a bad argument, a provider that blinked, a spell used out of order — costs the whole answer: the stream stops mid-way, whatever had already been drawn is abandoned, and the person is left with a trace id instead of a reply. The model is the one that can work around a failed call, and the one that can say what went wrong, so it is told.

ToolFailed rather than ModelRetry: a call that failed for reasons the model did not cause is not a call to make again, and a retry spends a budget that ends the run when it runs out — which is the same lost turn, one attempt later. A failure it can read leaves it free to try another way or explain. Runs stay bounded by Inkling's configured UsageLimits.request_limit, not by how many times one tool may break.

Source code in octomate/capabilities/tools.py

@dataclass
class ToolFailureCapability(AbstractCapability[None]):
    """Report a failed tool to the model instead of ending the turn.

    Pydantic AI ends a run when a tool raises anything but its own control-flow
    signals, so one broken call — a bad argument, a provider that blinked, a
    spell used out of order — costs the whole answer: the stream stops mid-way,
    whatever had already been drawn is abandoned, and the person is left with a
    trace id instead of a reply. The model is the one that can work around a
    failed call, and the one that can say what went wrong, so it is told.

    `ToolFailed` rather than `ModelRetry`: a call that failed for reasons the
    model did not cause is not a call to make again, and a retry spends a budget
    that ends the run when it runs out — which is the same lost turn, one attempt
    later. A failure it can read leaves it free to try another way or explain.
    Runs stay bounded by Inkling's configured `UsageLimits.request_limit`, not by
    how many times one tool may break.
    """

    async def on_tool_execute_error(
        self,
        ctx: RunContext[None],
        *,
        call: ToolCallPart,
        tool_def: ToolDefinition,
        args: ValidatedToolArgs,
        error: Exception,
    ) -> Never:
        # The model is told a sentence; the traceback is only ever seen here, so it
        # carries the exception itself and the ids to find the run it belongs to.
        react_logfire.exception(
            "tool {tool_name} failed; reported to the model",
            tool_name=call.tool_name,
            tool_call_id=call.tool_call_id,
            run_id=ctx.run_id,
            _exc_info=error,
        )
        raise ToolFailed(f"`{call.tool_name}` failed: {error}") from error

on_tool_execute_error async

on_tool_execute_error(
    ctx: RunContext[None],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
    error: Exception,
) -> Never

Source code in octomate/capabilities/tools.py

async def on_tool_execute_error(
    self,
    ctx: RunContext[None],
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
    error: Exception,
) -> Never:
    # The model is told a sentence; the traceback is only ever seen here, so it
    # carries the exception itself and the ids to find the run it belongs to.
    react_logfire.exception(
        "tool {tool_name} failed; reported to the model",
        tool_name=call.tool_name,
        tool_call_id=call.tool_call_id,
        run_id=ctx.run_id,
        _exc_info=error,
    )
    raise ToolFailed(f"`{call.tool_name}` failed: {error}") from error