Agent Configuration

Data Sources

• context.<key> - Values from the middleware context dict. Set in your before middleware. Supports nested paths like context.user.role.
• input.<key> - Values from the connector's JSON payload. Only works when the incoming payload is valid JSON. Supports nested paths like input.metadata.tier. If the payload is not valid JSON, all input.* checks evaluate to false.

Operators

Conditions use Python expression syntax: and, or, not for logic, ==, !=, >, <, >=, <= for comparisons, and parentheses for grouping. String literals use single quotes.

Setting Context in Middleware

async def before(content: dict, context: dict) -> dict:
    # Set context values that tool conditions can check
    context["multiply_allowed"] = True
    context["admin"] = content["parts"][0]["text"].startswith("/admin")
    return content

Queue Mode (default)

When a second event arrives for the same key while a run is active, it waits in a queue until the first run finishes, then processes automatically.

version: "1.0"

name: support-processor
type: llm
model: gemini/gemini-2.5-pro
description: "Processes support tickets one at a time per process"
system_prompt: "You are a support agent. Process the incoming ticket."

# Only one run per process_id at a time. Additional runs wait in queue.
concurrency:
  key: "process_id"
  on_conflict: queue

Drop Mode

When a second event arrives for the same key while a run is active, it is immediately cancelled. Use this when only the latest state matters and processing stale events would be wasteful.

version: "1.0"

name: notification-handler
type: tool
description: "Handles notifications, skipping duplicates"
tool_name: notifications.handle

# Drop duplicate runs for the same customer while one is active.
concurrency:
  key: "data.customer_id"
  on_conflict: drop

Key Extraction

The key field uses dot-notation to extract a value from the trigger payload. For example, if your Kafka message payload is {"data": {"customer_id": "abc"}}, use data.customer_id as the key. If the key path is not found in the payload, concurrency enforcement is skipped for that run.

Interaction with max_concurrent_runs

Both constraints apply simultaneously. For example, with max_concurrent_runs: 5 and concurrency.key: "process_id", up to 5 different process IDs can run in parallel, but only 1 run per process ID at a time.

Session Key

The key field determines how the agent identifies which session to use. It must start with one of two prefixes:

• context.<path> - Resolves from the middleware context dictionary. Set values in your before middleware. This is the recommended approach as it lets you derive a stable session key from any part of the request.
• input.<path> - Resolves from the raw connector payload (before middleware). Only works when the incoming payload is valid JSON. Supports nested paths like input.data.user_id.

Setting the Key via Middleware

async def before(content: dict, context: dict) -> dict:
    # Extract a unique identifier for this conversation
    context["chat_id"] = content["parts"][0]["text"].split(":")[0]
    return content

Using Input Keys

# When the connector payload is JSON like {"user_id": "u123", "message": "..."}
session:
  key: input.user_id

TTL (Time-to-Live)

The optional ttl field sets a session expiration in seconds. Sessions that have not been updated within the TTL period are considered expired and will be cleaned up automatically. If ttl is not set, sessions never expire. The minimum value is 60 seconds.

Managing Sessions

Active sessions can be viewed and deleted from the dashboard under Storage > Sessions. Sessions are scoped per environment.

Simple LLM Agent

A minimal LLM agent configuration:

version: "1.0"

name: assistant
type: llm  # Default type, can be omitted
model: gemini/gemini-2.5-pro  # Provider prefix required
description: "A helpful general-purpose assistant"
system_prompt: "You are a helpful assistant. Be concise and accurate."

Full-Featured LLM Agent

An invoice processing agent with multiple tools and detailed instructions:

version: "1.0"

name: invoice-processor
type: llm
model: openai/gpt-5.2  # Or: gemini/gemini-2.5-pro, ...
fallback_model: gemini/gemini-2.5-pro  # Used when primary model is unavailable
description: "Extracts data from invoices and validates totals"
system_prompt: |
  You are an expert accountant specializing in invoice processing.
  
  Your responsibilities:
  1. Extract all relevant fields from invoices (vendor, date, line items, totals)
  2. Use the calculator tool to verify mathematical accuracy
  3. Flag any discrepancies between line items and totals
  4. Format extracted data in a structured JSON format
  
  Always double-check calculations before confirming totals are correct.

max_concurrent_runs: 10
max_iterations: 50        # Stop after 50 iterations to limit cost
temperature: 0.7
reasoning: true           # Capture model reasoning in traces
reasoning_budget: 8192    # Max tokens for reasoning
retry_options:
  attempts: 5
  max_delay: 60
  # rerun_middleware: true
tools:
  - calculator.add
  - calculator.multiply
  - pdf.extract_text
  - validation.check_totals
guardrails:
  input:
    - type: prompt_injection
      mode: block
    - type: pii
      mode: redact
  output:
    - type: moderation
      mode: block
    - type: system_prompt_leakage
      mode: block

Document Processing Pipeline

A sequential agent that chains multiple agents together:

version: "1.0"

name: document-pipeline
type: sequential
description: "Processes documents through extraction and validation"

# Agents execute in order, each receiving the previous agent's output
agents:
  - assistant         # First: extracts key information
  - invoice-processor # Then: validates and processes the data

Direct Tool Execution

A tool agent that executes a calculator function directly:

version: "1.0"

name: tax-calculator
type: tool
description: "Calculates tax directly using the calculator tool"

# Executes this tool directly with the incoming payload
tool_name: calculator.calculate_tax

Chatbot with Session Memory

An agent that maintains conversation history per chat, with sessions expiring after 24 hours of inactivity:

version: "1.0"

name: support-bot
type: llm
model: gemini/gemini-2.5-pro
description: "A support chatbot that remembers conversation history"
system_prompt: |
  You are a helpful support agent. Use the conversation history
  to provide contextual responses.

# Maintain persistent sessions keyed by chat ID from middleware
session:
  key: context.chat_id
  ttl: 86400  # Sessions expire after 24 hours of inactivity

Setting the Session Key in Middleware

Use a before middleware to extract a stable identifier and set it in the context:

async def before(content: dict, context: dict) -> dict:
    # Extract a unique identifier for this conversation
    context["chat_id"] = content["parts"][0]["text"].split(":")[0]
    return content