Core concepts

This page explains the building blocks of an SDK workflow and how RunLLM executes them.

Workflows and runs

A workflow is what you publish with Client.publish(). It has a name, an entrypoint, an optional list of additional tasks, and an optional static config.

A workflow run is one execution of a workflow for a single conversation. The first event that matches the entrypoint's listeners in a new conversation (for example, a Slack thread or a chat widget session) starts a run. Later events in that same conversation go to the same run.

At any time, a run is in exactly one task:

A run starts in the entrypoint. While the run is still in the entrypoint, every new event in the conversation that matches the entrypoint's listeners calls the entrypoint again.
When a task calls agent.listen(), the run moves to a new task. From then on, only events on the surfaces passed to listen() are routed to the run, and they call the new task.

Entrypoints and tasks

Workflow functions are regular Python functions wrapped with one of two decorators from runllm.decorators.

`@entrypoint`

Marks the function that starts every workflow run, and declares which listeners trigger it.

from runllm import Agent, Event, SlackListener
from runllm.decorators import entrypoint


@entrypoint(listeners=[SlackListener(team_id="T0123456789")])
def on_question(agent: Agent, event: Event) -> None:
    ...

Argument	Type	Description
listeners	`List[Listener]`	Required, keyword-only. The listeners that start this workflow.
name	`str`	Optional. The task name. Defaults to the function name.

`@task`

Marks a follow-up function that a run can transition to with agent.listen().

from runllm.decorators import task


@task()
def follow_up(agent: Agent, event: Event, user_thread: SlackThread) -> None:
    ...

Argument	Type	Description
name	`str`	Optional, keyword-only. The task name. Defaults to the function name.

Pass every task to publish()

Tasks aren't discovered automatically. Pass every task your workflow can transition to in Client.publish(..., tasks=[...]), or the run will fail when it tries to transition to a task that wasn't published. Entrypoint and task names must be unique within a workflow.

Function signature

RunLLM calls every entrypoint and task with positional arguments in this order:

def my_function(agent, event, config, *surfaces): ...

Position	Type	When it's passed
`agent`	`Agent`	Always. Use it to take actions.
`event`	`Event`	Always. The event that triggered this call.
`config`	`Dict[str, Any]`	Only if you published the workflow with a `config`. See Static config.
`*surfaces`	`ChatSurface`	Only for tasks reached via `agent.listen()`. The surfaces passed to `listen(on=[...])`, in the same order.

For example, if a workflow is published with a config and a task calls agent.listen(on=[user_thread, internal_thread], handler=follow_up), then follow_up is called as follow_up(agent, event, config, user_thread, internal_thread).

Events and conversations

Every call receives an Event describing what happened:

event.conversation: the Conversation that the event happened in. Its new_message field is the user message that triggered the event, surface is where the conversation lives, and tags lists the tags already applied to it.
event.user: metadata about the user who triggered the event. event.user.email is currently only populated for Slack.
event.trigger: the trigger that matched.

Most agent actions take the Conversation (or its surface) as input:

convo = event.conversation
answer = agent.answer(convo)
agent.send_to_slack_thread(answer, to=convo.surface)

Listeners

Listeners are attached to the entrypoint and decide which new conversations start a workflow run. See the Types reference for every field.

Listener	Starts a run when
`SlackListener(team_id=...)`	A Slack message in the workspace matches the listener's triggers. By default, the trigger is a `Mention` of the RunLLM bot in any channel the bot has access to.
`WidgetListener(domain=...)`	A user sends a message in the RunLLM chat widget on the given domain.
`ZendeskListener(subdomain=..., trigger=...)`	A Zendesk event in the given subdomain matches the trigger, such as a new ticket or a new comment.

Use SlackChannel to scope a Slack listener to specific channels, each with its own triggers:

from runllm import ChannelMessage, Emoji, SlackChannel, SlackListener

SlackListener(
    team_id="T0123456789",
    channels=[
        # Respond to every top-level message in #support.
        SlackChannel(channel_id="C0SUPPORT01", trigger=ChannelMessage()),
        # Only start when someone reacts with :ticket: in #eng.
        SlackChannel(channel_id="C0ENGINEER1", trigger=Emoji(shortcode="ticket")),
    ],
)

Triggers

Triggers are the conditions that fire an event. You use them in listeners, and in agent.listen(..., triggers=[...]) when moving a run to a new task.

Trigger	Surfaces	Fires when
`Mention()`	Slack	A message mentions the RunLLM bot.
`ChannelMessage()`	Slack	Someone posts a top-level message in the channel. Thread replies don't count, and the bot doesn't need to be mentioned.
`Emoji(shortcode=...)`	Slack	Someone reacts with the given emoji on the last message in the conversation. Use the shortcode without colons (`"ticket"` for `:ticket:`). Reactions on user replies are ignored unless you set `exclude_replies=False`.
`ConvoMessage()`	Slack, chat widget	A new message is posted in an existing conversation, such as a reply in a Slack thread.
`TicketComment()`	Zendesk	A comment is added to a ticket.
`TicketCreated()`	Zendesk	A new ticket is created.

Surfaces

A surface is a place where a conversation happens and where the agent can send messages. Every surface has a session_id that identifies the conversation in RunLLM.

Surface	Created by	Send with
`SlackThread`	An incoming Slack event, or `agent.create_or_get_slack_thread()`	`agent.send_to_slack_thread()`
`ZendeskTicket`	An incoming Zendesk event, or `agent.create_zendesk_ticket()`	`agent.send_to_zendesk_ticket()`
`ChatWidget`	An incoming chat widget message	`agent.send_to_chat_widget()`

When a task is triggered by one of the surfaces it's listening on, event.conversation.surface is the same object as the matching positional surface argument, so you can check which surface fired with is.

Surfaces are live objects: when an action changes a surface (for example, the first send to a new Slack thread assigns its thread_ts, or update_zendesk_ticket() changes a ticket's status), the SDK updates the object in place.

Moving between tasks

Call agent.listen() to move the run to another task and wait for new events on one or more surfaces. A common pattern is escalating a user's question to an internal channel, then relaying the reply back to the user:

from runllm import (
    Agent,
    AnswerCategory,
    Client,
    ConvoMessage,
    Event,
    SlackListener,
    SlackThread,
)
from runllm.decorators import entrypoint, task

TEAM_ID = "T0123456789"
ESCALATION_CHANNEL = "C0ESCALATE1"


@task()
def relay_reply(
    agent: Agent,
    event: Event,
    user_thread: SlackThread,
    internal_thread: SlackThread,
) -> None:
    # Only relay messages posted by the support team in the internal thread.
    if event.conversation.surface is not internal_thread:
        return
    agent.send_to_slack_thread(event.conversation.new_message.text, to=user_thread)


@entrypoint(listeners=[SlackListener(team_id=TEAM_ID)])
def triage(agent: Agent, event: Event) -> None:
    convo = event.conversation
    answer = agent.answer(convo)
    agent.send_to_slack_thread(answer, to=convo.surface)

    if answer.category != AnswerCategory.ANSWERED:
        internal_thread = agent.create_or_get_slack_thread(
            session_id=convo.session_id,
            team_id=TEAM_ID,
            channel_id=ESCALATION_CHANNEL,
        )
        agent.send_to_slack_thread(
            f"Escalated question: {convo.new_message.text}",
            to=internal_thread,
            enable_feedback=False,
        )
        agent.listen(
            on=[convo.surface, internal_thread],
            handler=relay_reply,
            triggers=[ConvoMessage(), ConvoMessage()],
        )

Publish both functions:

Client().publish(name="escalation", entrypoint=triage, tasks=[relay_reply])

Buttons and suspended execution

Slack messages can include buttons (see send_to_slack_thread()). When you send buttons, the SDK suspends the task: your function stops at that line and the process exits cleanly. When the user clicks a button, RunLLM runs the same task again from the top with the same event. Every action your function already made is replayed from a cache instead of being re-executed, and the send_to_slack_thread() call that sent the buttons returns a ButtonClicked with the ID of the clicked button.

This has two consequences for code that runs before a button:

Keep it deterministic. The replay must make the same agent.* calls in the same order. If the sequence changes, the replay fails with a mismatch error.
Avoid side effects outside the agent. Only agent.* calls are cached. Anything else, such as calls to your own APIs, runs again on replay.

If the run has already moved to another task when a button is clicked, the click is ignored.

Static config

Pass a JSON-serializable config to Client.publish() to give every call in the workflow read-only settings, without hard-coding them into your functions:

@entrypoint(listeners=[SlackListener(team_id="T0123456789")])
def on_question(agent: Agent, event: Event, config: dict) -> None:
    ticket = agent.create_zendesk_ticket(
        subdomain=config["zendesk_subdomain"],
        conversation=event.conversation,
    )


Client().publish(
    name="support",
    entrypoint=on_question,
    config={"zendesk_subdomain": "acme"},
)

When you publish with a config, every entrypoint and task in the workflow receives it as the third argument.

Persistent state

Use agent.save_state() and agent.read_state() to store JSON-serializable values that persist across tasks and runs. State is scoped to the workflow, so all runs of the same workflow share the same keys.

count = agent.read_state("escalations") or 0
agent.save_state({"escalations": count + 1})

Workflows and runs​

Entrypoints and tasks​

@entrypoint​

@task​

Function signature​

Events and conversations​

Listeners​

Triggers​

Surfaces​

Moving between tasks​

Buttons and suspended execution​

Static config​

Persistent state​