Scaffold a New AI Agent

Generate all boilerplate for a new AI agent. The agent name/description should be provided via $ARGUMENTS.

Before You Start

Read the agent scope guide: packages/agent/CLAUDE.md

Study existing agents for reference patterns:

• Minimal reference: packages/agent/playground/minimal_workflow/simple_workflow/simple_agent.py
• Smallest chat blueprint (capabilities): packages/agent/swiss_ai_hub/agent/agents/llm_wrapping_agent/llm_wrapping_agent.py
• Production reference: packages/agent/swiss_ai_hub/agent/agents/rag_agent/rag_agent.py
• Pattern index: packages/agent/playground/minimal_workflow/ (20+ self-contained examples)

Agent classes hold steps only. A class derived from Agent defines @step methods and class attributes, nothing else — helpers go into standalone classes the steps import (or dispatcher-injected dependencies), never onto self. swiss_ai_hub/agent/agents/tests/test_agent_classes_hold_steps_only.py fails otherwise.

Conversational agents (chat UI) use capabilities, not hand-written steps. Title, follow-ups, meta-question gate, query condensing, memory, attached files, # knowledge references, tool loops and the final stop come from Conversation, Memory, AttachedFiles, Knowledge and ToolLoop under packages/agent/swiss_ai_hub/agent/capabilities/ — see "Step 4b" below.

Architecture & Mental Model

Agents are Dispatchable Workflows — directed acyclic graphs where nodes are @step methods and edges are typed Events. The framework is a custom workflow engine — not LlamaIndex workflows.

Fundamental invariant: Steps declare data requirements, not execution order. The dispatcher decides when to execute each step based on which events are available.

Inheritance chain: DispatchableWorkflow → Agent → your concrete agent

Key design properties:

• Stateless: Each step gets a fresh agent() instance. Never use self for state.
• Distributed: Consecutive steps may run on different servers via NATS/JetStream load balancing.
• Event-driven: All inter-step communication is through typed events. No shared memory, no direct calls.
• Data-injected: Step parameters are resolved by the dispatcher via dependency injection (type annotation → value).

Additional invariant properties:

• Any step can depend on any event, including the original StartEvent, regardless of how many steps have executed since. Events persist until run completion (Rule R6).
• Parallel execution is automatic: Steps with independent dependencies execute concurrently. The workflow is a dependency graph, not a sequence.
• Multiple steps for the same run may execute in parallel if their dependencies are independently satisfied.

Critical difference — think data dependencies, not control flow:

# WRONG mental model (imperative, top-down):
# "First do A, then B, then C"
async def run(self):
    a = await self.step_a()
    b = await self.step_b(a)
    c = await self.step_c(b)

# CORRECT mental model (declarative, bottom-up):
# "C needs B's output. B needs A's output. A needs the start event."
@step()
async def step_a(self, event: StartEvent) -> EventA: ...

@step()
async def step_b(self, event: EventA) -> EventB: ...

@step()
async def step_c(self, event: EventB) -> StopEvent: ...

Avoid pass-through pollution — each event should contain only the data it semantically represents:

# WRONG (top-down thinking, passing data forward):
@step()
async def retrieve(self, event: UserMessageEvent) -> RetrieveEvent:
    nodes = await retriever.retrieve(event.user_query)
    return RetrieveEvent(nodes=nodes, user_query=event.user_query)  # Passing query forward!

@step()
async def respond(self, event: RetrieveEvent) -> StopEvent:
    return await generate(event.user_query, event.nodes)  # Using passed-through data

# CORRECT (bottom-up thinking, direct dependencies):
@step()
async def retrieve(self, event: UserMessageEvent) -> RetrieveEvent:
    nodes = await retriever.retrieve(event.user_query)
    return RetrieveEvent(nodes=nodes)  # Only retrieval-specific data

@step()
async def respond(
    self,
    retrieve_event: RetrieveEvent,
    user_event: UserMessageEvent,  # Direct dependency on original event
) -> StopEvent:
    return await generate(user_event.user_query, retrieve_event.nodes)

Design approach: Sketch top-down to understand logical flow, then refine bottom-up to identify true data dependencies. For each step ask: What is the minimal set of data this step requires?

Source: packages/core/swiss_ai_hub/core/workflow/dispatchable_workflow.py, packages/agent/swiss_ai_hub/agent/agents/agent.py

Step 1: Choose Your Pattern

Select the pattern that matches your agent's workflow. Each maps to a playground example.

Pattern 1: Linear Pipeline

Steps execute in sequence, each consuming the previous step's output.

@step()
async def start_step(self, event: UserMessageEvent) -> EventA: ...
@step()
async def process_step(self, event: EventA) -> EventB: ...
@step()
async def end_step(self, event: EventB) -> StopEvent: ...

Playground: playground/minimal_workflow/simple_workflow/

Pattern 2: Conditional Branching

A step returns one of several event types based on a condition.

@step()
async def start_step(self, event: StartEvent) -> AboveEvent | BelowEvent:
    if condition:
        return AboveEvent()
    return BelowEvent()

@step()
async def end_step(self, event: AboveEvent | BelowEvent) -> StopEvent: ...

Playground: playground/minimal_workflow/conditional_workflow/

Pattern 3: Fan-Out / Fan-In

Parallel processing: one step produces multiple events, another collects all results.

@step()
async def fan_out(self, _: StartEvent) -> list[TaskEvent]:
    return [TaskEvent(task=t) for t in tasks]

@step()
async def process(self, event: TaskEvent) -> ResultEvent:
    return ResultEvent(result=process(event.task))  # Runs once per TaskEvent

@step()
async def fan_in(self, results: FixedList(ResultEvent, N)) -> StopEvent:
    # Waits for exactly N results, then fires once
    return StopEvent(combined=[r.result for r in results])

Key: Use FixedList(EventType, N) when the count is known at compile time.

Playground: playground/minimal_workflow/fan_out_workflow/

Pattern 4: Precondition Sync

Wait for a dynamic number of events using a precondition function.

def ensure_enough_events(events: list[ParallelEvent], config: MyConfig) -> bool:
    return len(events) == config.expected_count

@step(precondition=ensure_enough_events)
async def collect_step(self, events: list[ParallelEvent]) -> StopEvent: ...

Key: The precondition function uses the same DI system as @step — it can receive events, config, context. The precondition re-evaluates on each new event arrival until it returns True.

Playground: playground/minimal_workflow/precondition_workflow/

Pattern 5: Bounded Loop

Iterative processing with a counter-based exit condition.

@step()
async def start_step(self, event: UserMessageEvent, ctx: RunContext) -> BeginEvent:
    await ctx.set("iteration", 0)
    return BeginEvent()

@step()
async def process_step(self, event: BeginEvent) -> ProcessedEvent: ...

@step()
async def decision_step(self, event: ProcessedEvent, ctx: RunContext) -> DoneEvent | BeginEvent:
    count = await ctx.get("iteration", 0) + 1
    await ctx.set("iteration", count)
    if count >= MAX_ITERATIONS or event.is_good_enough:
        return DoneEvent()
    return BeginEvent()  # Loop back

@step()
async def end_step(self, event: DoneEvent) -> StopEvent: ...

Key: Use RunContext for the loop counter. Set max_executions_per_run as a safety limit.

Playground: playground/minimal_workflow/bounded_loop/

Pattern 6: Human-in-the-Loop (HITL)

Pause workflow to collect human input, then resume. Four HITL subtypes:

@step()
async def start_step(self, event: StartEvent) -> HumanInTheLoopInput.request:
    return HumanInTheLoopInput.invoke(question="Please enter your feedback:")

@step()
async def end_step(self, event: HumanInTheLoopInput.response) -> StopEvent:
    user_data = event.response  # User's text response
    return StopEvent()

Source: packages/core/swiss_ai_hub/core/events/agent/hitl/

Multiple HITL Interactions

For workflows requiring multiple human interactions, create distinct subclasses. The dispatcher differentiates steps by event type—using the same base type for multiple interactions causes ambiguity.

Step 1: Define custom HITL event pairs

# events/first_step_human_in_the_loop.py
from swiss_ai_hub.core.events.agent import (
    HumanInTheLoopInput,
    HumanInTheLoopInputRequestEvent,
    HumanInTheLoopInputResponseEvent,
)


class FirstStepHumanInTheLoopRequestEvent(HumanInTheLoopInputRequestEvent):
    pass


class FirstStepHumanInTheLoopResponseEvent(HumanInTheLoopInputResponseEvent):
    pass


class FirstStepHumanInTheLoop(HumanInTheLoopInput):
    request = FirstStepHumanInTheLoopRequestEvent
    response = FirstStepHumanInTheLoopResponseEvent

# events/second_step_human_in_the_loop.py
from swiss_ai_hub.core.events.agent import (
    HumanInTheLoopInput,
    HumanInTheLoopInputRequestEvent,
    HumanInTheLoopInputResponseEvent,
)


class SecondStepHumanInTheLoopRequestEvent(HumanInTheLoopInputRequestEvent):
    pass


class SecondStepHumanInTheLoopResponseEvent(HumanInTheLoopInputResponseEvent):
    pass


class SecondStepHumanInTheLoop(HumanInTheLoopInput):
    request = SecondStepHumanInTheLoopRequestEvent
    response = SecondStepHumanInTheLoopResponseEvent

Step 2: Use distinct types in the workflow

from swiss_ai_hub.core.events.agent import StartEvent, StopEvent

from swiss_ai_hub.agent.agents.agent import Agent
from swiss_ai_hub.agent.workflow.decorators.step import step

from .events.first_step_human_in_the_loop import FirstStepHumanInTheLoop
from .events.second_step_human_in_the_loop import SecondStepHumanInTheLoop


class MultistepHumanInTheLoopAgent(Agent):
    @step()
    async def start_step(self, event: StartEvent) -> FirstStepHumanInTheLoop.request:
        return FirstStepHumanInTheLoop.invoke(question="Shall I continue?")

    @step()
    async def second_hitl(
        self, event: FirstStepHumanInTheLoop.response
    ) -> SecondStepHumanInTheLoop.request:
        print(f"First response: {event.response}")
        return SecondStepHumanInTheLoop.invoke(question="Are you sure?")

    @step()
    async def end_step(self, event: SecondStepHumanInTheLoop.response) -> StopEvent:
        print(f"Second response: {event.response}")
        return StopEvent()

Dynamic HITL Type Selection

When the HITL type depends on runtime conditions, use union return types:

from swiss_ai_hub.core.events.agent import (
    HumanInTheLoopChat,
    HumanInTheLoopConfirmation,
    HumanInTheLoopInput,
)

class HitlDemoAgent(Agent):
    @step()
    async def select_hitl_type(
        self, event: UserMessageEvent
    ) -> HumanInTheLoopInput.request | HumanInTheLoopConfirmation.request | HumanInTheLoopChat.request:
        choice = event.user_query.lower()
        if "confirmation" in choice:
            return HumanInTheLoopConfirmation.invoke("Do you confirm this action?")
        elif "chat" in choice:
            return HumanInTheLoopChat.invoke("What is your response?")
        else:
            return HumanInTheLoopInput.invoke("Please enter your text input:")

    @step()
    async def handle_response(
        self,
        event: HumanInTheLoopInput.response | HumanInTheLoopConfirmation.response | HumanInTheLoopChat.response,
    ) -> StopEvent:
        if isinstance(event, HumanInTheLoopConfirmation.response):
            result = f"Confirmation: {'Yes' if event.response else 'No'}"
        else:
            result = f"Response: {event.response}"
        return StopEvent()

Pattern 7: Bot-in-the-Loop (BITL)

Delegate to a human via Teams/Slack bot channel, then resume with their response.

Channel Configuration

BITL requires platform-specific configuration:

Microsoft Teams:

from swiss_ai_hub.core.events.agent import TeamsConfig

teams_config = TeamsConfig(
    channel_id="19:abc123@thread.tacv2",
    tenant_id="12345678-1234-1234-1234-123456789abc",
    bot_id="87654321-4321-4321-4321-cba987654321",
)

Slack:

from swiss_ai_hub.core.events.agent import SlackConfig

slack_config = SlackConfig(
    channel_id="C0123456789",
    service_url="https://slack.botframework.com",
)

Basic Usage

from swiss_ai_hub.core.events.agent import BotInTheLoop

class BotInTheLoopAgent(Agent):
    @step()
    async def request_approval(self, start_event: MyStartEvent) -> BotInTheLoop.request:
        return BotInTheLoop.invoke(
            user=start_event.user,
            question="Should the agent proceed with the deployment?",
            channel_config=start_event.channel_config,  # TeamsConfig or SlackConfig
        )

    @step()
    async def handle_response(self, event: BotInTheLoop.response) -> StopEvent:
        answer = event.response
        if event.responder:
            print(f"Answered by: {event.responder.user_name}")
        return StopEvent()

Iterative Conversations

BITL supports multi-turn conversations by returning another BotInTheLoop.request:

@step()
async def handle_response(
    self, event: BotInTheLoop.response
) -> BotInTheLoop.request | StopEvent:
    if event.response.lower() == "yes":
        return StopEvent()
    else:
        return BotInTheLoop.invoke(
            user=event.request_event.user,
            question="What about now? Ready to proceed?",
            channel_config=event.request_event.channel_config,
        )

Response Event Structure

Responder Information:

BITL vs HITL

Requires: A bot agent configured with the appropriate channel. See /bot-framework for bot setup.

Playground: playground/agent/bot_in_the_loop_agent/

Pattern 8: Agent-in-the-Loop (AITL)

Delegate to another agent, then resume with their result.

@step()
async def start_step(self, event: UserMessageEvent) -> AgentInTheLoop.request:
    return AgentInTheLoop.invoke(
        agent_class="WorkerAgent",
        agent_id="worker-1",
        start_event=UserMessageEvent(messages=event.messages, user=event.user, locale=event.locale),
    )

@step()
async def handle_response(self, response: AgentInTheLoop.response) -> StopEvent:
    worker_result = response.stop_event
    return StopEvent()

@step()
async def handle_exception(self, exception: AgentInTheLoop.exception) -> StopEvent:
    return StopEvent(error=str(exception.exception_event))

Key: Always handle both .response and .exception from the delegated agent.

Playground: playground/minimal_workflow/agent_in_the_loop_workflow/

Step 2: Create Directory Structure

Extract the agent name from $ARGUMENTS. Convert to CamelCase for classes, snake_case for files and directories.

packages/agent/swiss_ai_hub/agent/agents/{agent_name}/
├── __init__.py                 # Exports {AgentName} and {AgentName}Config
├── {agent_name}.py             # Agent class (@step methods only)
├── configs/
│   └── {agent_name}_config.py  # AgentConfig subclass with form duality
├── events/
│   └── {event_name}.py         # One file per custom event (one class per file)
└── tests/
    ├── features/
    │   └── {agent_name}.feature  # BDD scenario
    └── test_{agent_name}.py      # Test implementation

packages/agent/app/{agent_name}/
├── main.py                     # Entry point with AgentRunner
└── Dockerfile

packages/agent/swiss_ai_hub/agent/i18n/translations/agent/
├── {agent_name}.de.yml
├── {agent_name}.en.yml
├── {agent_name}.fr.yml
└── {agent_name}.it.yml

Production agents also get a run-{agent-name} target in packages/agent/Makefile and a service in the compose template (infra/deployment/templates/docker-compose.yml.j2, then make generate-compose). Look at how llm_wrapping_agent is wired through app/llm_wrapping_agent/ for a complete example. Playground agents live under packages/agent/playground/ and use LocaleString instead of i18n files.

Naming conventions:

• One class per file, snake_case file name matches the class: my_agent.py contains class MyAgent
• Events: {action}_event.py — e.g., summary_generated_event.py contains SummaryGeneratedEvent
• Config: {agent_name}_config.py containing {AgentName}Config
• Helper logic that is not a step goes into its own class next to the agent, imported by the steps

Step 3: Define Events

Choosing the Right Base Event

UserMessageEvent is a chat UI contract. It is the canonical entry point for chat interfaces (OpenWebUI, Teams, Slack). Keep its payload minimal — every field added to it (or to a subclass) raises the bar for every chat client. If your agent needs a richer entry payload and the publisher is not a generic chat UI (e.g. a custom domain front-end or another agent delegating via AgentInTheLoop), subclass StartEvent directly and accept UserMessageEvent | YourStartEvent on the relevant steps.

Complete event selection guide (from the report):

Rule of thumb: If a step consumes it → ControlEvent. If only the UI needs it → DisplayEvent. If both → ControlAndDisplayEvent. Most custom agent events are ControlEvent.

Chat-UI agents do not wire self-awareness, titles or follow-ups by hand. The meta-question gate ("What can you do?"), the thread title and the follow-up questions belong to the Conversation capability (contextualize runs the gate and titles the thread, complete generates follow-ups and ends the run). See "Step 4b".

The Stop Event Constraint

No step may depend on StopEvent or any subclass as an input. When StopEvent is emitted, the run terminates.

# ILLEGAL: Depending on stop event
@step()
async def cleanup(self, stop: LLMStopEvent) -> CleanupEvent:
    ...  # Never executes

# CORRECT: Use non-stop intermediate event, then explicit stop
@step()
async def respond(self, event: Input) -> LLMEvent:  # Not LLMStopEvent
    return await displayer.display_llm_stream(..., as_stop_step=False)

@step()
async def cleanup(self, llm: LLMEvent) -> CleanupEvent:
    ...

@step(precondition=cleanup_complete)
async def finalize(self, cleanup: CleanupEvent) -> StopEvent:
    return StopEvent()

Custom Event Template

Create one file per event in agents/{agent_name}/events/:

# packages/agent/swiss_ai_hub/agent/agents/{agent_name}/events/{event_name}.py
from swiss_ai_hub.core.events.agent import ControlEvent


class {EventName}(ControlEvent):
    """Carries {description} from step X to step Y."""
    field_name: str
    another_field: list[str] = []

Events auto-register on import — no manual registration needed. See /nats-events for the full event hierarchy.

Events as Flow Carriers

Events serve two distinct purposes:

1. Data carriers: Transporting values between steps
2. Flow carriers: Controlling execution order independent of data

A step may depend on an event solely to ensure execution ordering:

class PathA(Event):
    pass  # No fields - pure flow control

@step()
async def handle_path_a(self, _: PathA, original: StartEvent) -> StopEvent:
    # Underscore signals: "I need this event for flow control, not data"
    process(original.data)
    return StopEvent()

The _: EventType convention indicates dependency on an event's existence rather than its contents. Essential for conditional branching, sequencing without data coupling, and synchronization barriers.

Step 4: Create Agent Class

# packages/agent/swiss_ai_hub/agent/agents/{agent_name}/{agent_name}.py
from typing import ClassVar

from swiss_ai_hub.core.events.agent import StopEvent, UserMessageEvent

from swiss_ai_hub.agent.agents.agent import Agent
from swiss_ai_hub.agent.agents.{agent_name}.events.{event_name} import {EventName}
from swiss_ai_hub.agent.i18n.agent_locale_string import AgentLocaleString
from swiss_ai_hub.agent.workflow.decorators.step import step


class {AgentName}(Agent):
    name: ClassVar[AgentLocaleString] = AgentLocaleString.from_i18n_path(
        "agent.{agent_name}.metadata.name"
    )
    description: ClassVar[AgentLocaleString] = AgentLocaleString.from_i18n_path(
        "agent.{agent_name}.metadata.description"
    )
    icon: ClassVar[str] = "mage:robot"

    @step(
        name=AgentLocaleString.from_i18n_path("agent.{agent_name}.steps.start.name"),
        description=AgentLocaleString.from_i18n_path("agent.{agent_name}.steps.start.description"),
        icon="mage:play",
    )
    async def start_step(self, event: UserMessageEvent) -> {EventName}:
        return {EventName}(field_name=event.user_query)

    @step(
        name=AgentLocaleString.from_i18n_path("agent.{agent_name}.steps.end.name"),
        icon="mage:check",
    )
    async def end_step(self, event: {EventName}) -> StopEvent:
        return StopEvent()

@step Decorator Parameters

Source: packages/agent/swiss_ai_hub/agent/workflow/decorators/step.py

Step Return Types

Step Parameter Types

The dispatcher resolves parameters by type annotation. Declare what you need:

Source: AgentDispatcher._get_parameter_value() in packages/agent/swiss_ai_hub/agent/dispatchers/agent_dispatcher.py

Event Resolution Strategy

When binding events to parameters:

1. Fixed Collection: FixedList(Event, N) blocks until exactly N events available, then returns all N
2. Unbounded List: list[Event] returns all events of that type currently available; step re-executes on each new arrival
3. Single Instance:
   • If the triggering event matches the parameter type, that instance is used
   • Otherwise, the most recently created event of that type is used

Ordering Guarantee: Events in list parameters are ordered by arrival time.

The Six Execution Rules

For debugging execution issues, see /debug-agent.

Step 4b: Chat Blueprints Use Capabilities

A conversational agent (chat UI, bots) is assembled from capabilities (packages/agent/swiss_ai_hub/agent/capabilities/), not from hand-written retrieval, memory, title or stop steps. A step returns a capability's request event through its typed helper; a later step declares the result event as a parameter. The reference is agents/llm_wrapping_agent/llm_wrapping_agent.py; the full contract is in packages/agent/CLAUDE.md ("Capabilities").

Typical spine of a chat blueprint (each line is one @step):

1. limit_chat_history_step(UserMessageEvent) -> Conversation.ContextualizeRequest (Conversation.contextualize(...))
2. gather_context_step(Conversation.Contextualized, UserMessageEvent) -> list[Memory.RecallRequest | AttachedFiles.ReadRequest | Knowledge.SearchRequest] (fan-out by returning a list)
3. assemble_prompt_step(Conversation.Contextualized, Memory.Recalled, AttachedFiles.Contents, Knowledge.Searched) -> Conversation.ComposeRequest
4. respond_step(Conversation.Composed, Conversation.Contextualized, ...) -> list[MemoryStorageRequestedEvent | Conversation.CompleteRequest]: stream with displayer.display_llm_stream(..., as_stop_step=False), then return Memory.remember(...) first and Conversation.complete(answer=...) last

Rules:

• The config class inherits the mixins of every capability it calls (class MyConfig(MemoryFields, ConversationFields, AgentConfig)) and spreads their *_form_elements() in as_form() (see LLMWrappingAgentConfig). The runner refuses to start otherwise.
• Annotate steps with the scoped names (Conversation.Composed, Memory.Recalled); requests carry a Request suffix.
• Do not write your own steps for the meta-question gate, query condensing, title, follow-ups, memory storage or the terminal stop event.
• A model that picks its own tools (Universal Agent pattern): declare tools = ToolLoop.over(Knowledge, AttachedFiles, Memory, MyToolSpec) on the class and call MyAgent.tools.run(history); capabilities are offered as tools and BaseToolSpec classes (LlamaIndex) add custom ones. The config then also lists ToolLoopFields, the *Fields mixin of every capability in the tool set, and KnowledgeToolFields when Knowledge is offered. Example: playground/minimal_workflow/tool_loop_workflow/.
• All validation is in Agent.validate_workflow (run by AgentRunner); capabilities/tests/test_capability_composition.py shows what it rejects.
• Protocol events of capabilities and the tool loop are ControlAndDisplayEvents with their own frontend component (see /scaffold-event-display), except the display-only ToolLoopStatusEvent and ToolLoopCondensedEvent.

Step 5: Create Config

# packages/agent/swiss_ai_hub/agent/agents/{agent_name}/configs/{agent_name}_config.py
from typing import Annotated, Self

from pydantic import Field

from swiss_ai_hub.core.agents import AgentConfig
from swiss_ai_hub.core.form import InputNumber, InputText, ModelSelect
from swiss_ai_hub.core.form.constraints import Ge, Le
from swiss_ai_hub.core.i18n import LocaleString


class {AgentName}Config(AgentConfig):
    model_name: Annotated[str | ModelSelect, Field(description="LLM model")] = "gpt-4o"
    temperature: Annotated[float | InputNumber, Field(description="LLM temperature"), Ge(0.0), Le(2.0)] = 0.7
    system_prompt: Annotated[str | InputText, Field(description="System prompt")] = "You are a helpful assistant."

    @classmethod
    def as_form(cls) -> Self:
        base = AgentConfig.as_form()
        return cls(
            **base.model_dump(),
            model_name=ModelSelect(label=LocaleString(en="Model", de="Modell", fr="Modèle", it="Modello")),
            temperature=InputNumber(
                label=LocaleString(en="Temperature", de="Temperatur", fr="Température", it="Temperatura"),
                min=0.0,
                max=2.0,
                step=0.1,
            ),
            system_prompt=InputText(
                label=LocaleString(en="System Prompt", de="Systemprompt", fr="Prompt système", it="Prompt di sistema"),
            ),
        )

Form Duality Rules

• Field types: Always primitive_type | FormkitElement union (e.g., str | InputText)
• Constraints: Use Ge(), Le(), Gt(), Lt(), MinLen(), MaxLen(), Pattern() from swiss_ai_hub.core.form.constraints — NOT Pydantic's ge=, le=
• Configurable fields: Set to a FormkitElement in as_form() → editable in Admin UI
• Non-configurable fields: Set to a primitive in as_form() → deployment-fixed, baked in
• Labels: All labels must be LocaleString with de, en, fr, it

StepConfig

For step-specific configuration, nest a StepConfig subclass:

from swiss_ai_hub.core.agents import StepConfig

class MyStepConfig(StepConfig):
    threshold: Annotated[float | InputNumber, Field(description="Threshold")] = 0.5

class {AgentName}Config(AgentConfig):
    my_step_settings: MyStepConfig = MyStepConfig()

The dispatcher auto-extracts StepConfig fields and injects them into steps that declare the matching type.

FormKit Elements Reference

From packages/core/swiss_ai_hub/core/form/elements/:

• Input: InputText, InputNumber, Textarea, Password, InputMask, InputOtp
• Selection: Select, MultiSelect, CascadeSelect, Checkbox, ToggleSwitch, ToggleButton, RadioButton, SelectButton, Listbox
• Specialized: ModelSelect (LLM picker), AgentSelector, KnowledgeDatabaseSelector, VectorStoreInput, IconSelector, LocaleInput (multi-language), ColorPicker, DatePicker, Knob, Rating, Slider
• Layout: Group (auto-created from nested Form), Repeater (auto-created from list[Form])

For a chat blueprint, add the capability mixins (ConversationFields, MemoryFields, ...) as bases and spread their *_form_elements() into as_form() — see LLMWrappingAgentConfig.

Source: packages/core/swiss_ai_hub/core/agents/agent_config.py, packages/core/swiss_ai_hub/core/form/form.py

Step 6: Create Entry Point

# packages/agent/app/{agent_name}/main.py
# ruff: noqa: E402
from swiss_ai_hub.core.infrastructure import AihubInstrumentor  # isort: skip

AihubInstrumentor().instrument()

import asyncio

from swiss_ai_hub.core.infrastructure import AIHubSettings, enable_logging

from swiss_ai_hub.agent.agents.{agent_name} import {AgentName}, {AgentName}Config
from swiss_ai_hub.agent.runners import AgentRunner

enable_logging()


async def main():
    runner = AgentRunner(agent_type={AgentName}, agent_config={AgentName}Config.as_form())
    await runner.run_forever()


if __name__ == "__main__":
    print(AIHubSettings().startup_banner)
    asyncio.run(main())

Model on packages/agent/app/llm_wrapping_agent/main.py (it also passes templates= for agent profile templates, optional).

Source: packages/agent/swiss_ai_hub/agent/runners/agent_runner.py

Step 7: Add i18n

Create translation files in packages/agent/swiss_ai_hub/agent/i18n/translations/agent/. The file name is the path segment after agent., so no locale root key is needed:

# {agent_name}.en.yml
metadata:
  name: "{Agent Display Name}"
  description: "{Agent description for Admin UI}"
steps:
  start:
    name: "Start"
    description: "Receives user message and begins processing"
  end:
    name: "Finish"

Create matching files for de, fr, it locales with translated strings.

Translation lookup order: Local → Agent Scope → Library → English fallback

Usage in agent class:

name: ClassVar[AgentLocaleString] = AgentLocaleString.from_i18n_path("agent.{agent_name}.metadata.name")

Usage in step methods (via DI):

@step()
async def my_step(self, event: MyEvent, t: LocaleHandler) -> StopEvent:
    localized_text = t("agent.{agent_name}.some_key")
    return StopEvent()

Source: packages/agent/swiss_ai_hub/agent/i18n/agent_locale_string.py, packages/agent/swiss_ai_hub/agent/i18n/translations/agent/

Step 8: LLM Integration

Use EventDisplayer for streaming LLM output to the frontend. AgentConfig carries the LLMConfig (config.llm); cost_reporting_llm yields an LLM that reports its costs as LLMCostEvents:

from swiss_ai_hub.core.displayers import EventDisplayer
from swiss_ai_hub.core.events.agent import LLMEvent

@step()
async def llm_step(
    self,
    event: MyEvent,
    config: MyAgentConfig,
    displayer: EventDisplayer,
    user: UserIdentity | None = None,
) -> LLMEvent:
    messages = [
        ChatMessage(role=MessageRole.SYSTEM, content=config.system_prompt),
        ChatMessage(role=MessageRole.USER, content=event.message),
    ]
    async with config.llm.cost_reporting_llm(displayer, user=user) as llm:
        return await displayer.display_llm_stream(config.llm, llm, messages, as_stop_step=False)

Key: display_llm_stream() handles token-by-token streaming, <think> tag parsing, and ChunkEvent emission automatically. With as_stop_step=True it returns an LLMStopEvent that ends the run; a chat blueprint passes False and hands the answer to Conversation.complete(...). Pass tools= only inside the tool loop.

Display Methods

Source: packages/core/swiss_ai_hub/core/displayers/event_displayer.py, packages/core/swiss_ai_hub/core/displayers/stream/stream_processor.py

Step 9: Memory Integration

Chat blueprints: use the Memory capability (see "Step 4b"): Memory.recall(ctx.query) ahead of the prompt, Memory.remember(...) returned before Conversation.complete(...) so the storage request is published before the run tears down. The config needs MemoryFields; the profile decides which scopes are read and written.

Non-chat agents that talk to memory directly inject AgentMemory (built from the profile's own extraction model) and call search_user_memory / search_organization_memory / add_user_memory:

from swiss_ai_hub.core.generative_ai import AgentMemory

@step()
async def retrieve_memory(self, event: UserMessageEvent, memory: AgentMemory) -> MemoryEvent:
    user_memories = await memory.search_user_memory(query=event.user_query, user_id=event.user.id)
    return MemoryEvent(user_memories=user_memories)

Termination constraint: StopEvent must come AFTER memory storage — the store step produces an intermediate event and the stop step consumes it.

Playground: playground/minimal_workflow/user_memory_workflow/, playground/minimal_workflow/organization_memory_workflow/

Source: packages/core/swiss_ai_hub/core/generative_ai/memory/agent_memory.py

Step 10: Create Tests

Runner-backed tests need the Docker dev stack (NATS, API for config RPC). Stop any locally running agent of the same class first — it shares the test runner's NATS queue group and eats its events.

BDD Feature File

# packages/agent/swiss_ai_hub/agent/agents/{agent_name}/tests/features/{agent_name}.feature
Feature: {Agent Display Name}

  Scenario: Happy path
    Given a {AgentName} runner
    When the start event is sent with payload "test message"
    Then a StopEvent is present

Test Implementation

Model on packages/agent/playground/minimal_workflow/simple_workflow/tests/test_simple_agent.py:

# packages/agent/swiss_ai_hub/agent/agents/{agent_name}/tests/test_{agent_name}.py
from llama_index.core.base.llms.types import ChatMessage, MessageRole
from pytest_bdd import given, parsers, scenarios, then, when
from swiss_ai_hub.core.events.agent import UserMessageEvent
from swiss_ai_hub.core.i18n import LocaleString
from swiss_ai_hub.core.testing import async_test
from swiss_ai_hub.core.testing.auth_utils import fake_user

from swiss_ai_hub.agent.agents.{agent_name} import {AgentName}, {AgentName}Config
from swiss_ai_hub.agent.runners.agent_test_runner import AgentTestRunner

scenarios("./features/{agent_name}.feature")


@given("a {AgentName} runner", target_fixture="agent_runner")
def _():
    return AgentTestRunner(
        agent_type={AgentName},
        agent_config={AgentName}Config(agent_id="{agent_name}", name=LocaleString(en="{Agent Display Name}"),
                                       description=LocaleString(en="Test")),
    )


@when(parsers.parse('the start event is sent with payload "{payload}"'))
@async_test
async def _(agent_runner: AgentTestRunner, payload: str):
    async with agent_runner.test_run() as topic:
        await agent_runner.send_event_from_topic(
            start_event=UserMessageEvent(
                messages=[ChatMessage(content=payload, role=MessageRole.USER)], user=fake_user()
            ),
            topic=topic,
        )


@then("a StopEvent is present")
def _(agent_runner: AgentTestRunner):
    assert agent_runner.has_stop_event, "Agent did not receive stop event"

A blueprint that uses memory or other agents needs await agent_runner.ensure_dependent_agent_stream(...) inside the test_run() block (see rag_agent/tests/test_rag_agent.py).

Key Test Assertions

For AITL tests, use runner.ensure_dependent_agent_stream(agent_class).

Unit Testing (Direct Step Invocation)

Individual steps can be tested by calling them directly, bypassing the dispatcher (see llm_wrapping_agent/tests/test_llm_wrapping_input_too_large_guard.py):

result = await MyAgent().retrieve_step(event=event, memory=memory)

Helper classes the steps import are tested on their own. Capability composition (a blueprint that stalls or crashes) is pinned by capabilities/tests/test_capability_composition.py.

Source: packages/agent/swiss_ai_hub/agent/runners/agent_test_runner.py

Step 11: Register & Verify

Agent discovery is automatic — AgentRunner responds to AgentClassDiscoveryRequestEvent with the agent's metadata, form schema, event specs, and workflow graph. No manual registration needed.

Verify the agent is discoverable:

cd packages/agent && uv run python -c "from swiss_ai_hub.agent.agents.{agent_name} import {AgentName}; print({AgentName}.get_steps())"

Implementation Checklist

Before Coding

• Read packages/agent/CLAUDE.md
• Identified the pattern from the catalog above
• Studied the matching playground example
• Sketched the event DAG on paper (events as edges, steps as nodes)

During Coding

• Agent class extends Agent with name, description, icon as ClassVar[AgentLocaleString]
• All @step methods are async, use type annotations, return events
• No instance state on self and no helper methods on the class — state in RunContext / ThreadContext, helpers in imported classes
• Events inherit from the correct base class (see table above)
• One class per file, file name matches class name
• Config uses form duality pattern with as_form() classmethod
• Form constraints use Ge(), Le() etc. — not Pydantic's ge=, le=
• i18n translations in all 4 locales (de, en, fr, it)
• Entry point in app/{agent_name}/main.py; chat agent: config carries the capability mixins

After Coding

• Every event produced by a step is consumed by another step (no dead ends)
• Every execution path reaches StopEvent
• StopEvent is returned alone (not in a list with other events)
• Optional params have synchronization (precondition or max_executions_per_run)
• BDD tests pass: cd packages/agent && uv run pytest swiss_ai_hub/agent/agents/{agent_name} -v
• Agent is importable: uv run python -c "from swiss_ai_hub.agent.agents.{agent_name} import {AgentName}"

File Reference

Framework Files

Playground Patterns Index

Commands

make pr-ready    # Format + lint (also run by the stop hook)
make test        # Run pytest in the scope (from packages/agent or the root)

Cross-References

• Debugging agents: /debug-agent — execution semantics, race conditions, MCP-powered diagnostics
• Event infrastructure: /nats-events — event hierarchy, subject format, dispatcher architecture
• Event display components: /scaffold-event-display — frontend visualization for new event types
• Bot integration: /bot-framework — CompletionHandler, channel setup for BITL patterns
• Process orchestration: see packages/process/CLAUDE.md — multi-entity workflows (agents + humans + programs)