Xubb Agents Framework

A swarm of cheap, specialized LLM observers that watch a live conversation and surface the one insight worth saying. Most agents stay silent most of the time; that restraint is the product.

Version: 3.2.0 · Status: Beta, production-hardened (every documented contract is CI-gated; see docs/PROCESS.md)

📚 Docs index · 🔒 Security · 📝 Changelog · 🏛 Architecture

Note: This is a separate, standalone library, designed to be embedded in any host application that needs real-time conversation agents — a web server, a desktop app, a CLI tool.

📖 New here? Work down this page — install, run it offline, run it with a model, read the result — then the design guide for how to wield it well. The API reference and diagnostics are for when you are building against it.

Contents

Start here — What this is · Install · Run it now · With a model · Reading the result · Where to go next

Reference — Architecture · Core Concepts · Blackboard · Triggers · Agent Configuration · Agent Communication · Usage Guide · API Reference · Best Practices · Migration · Contributing

What this is

A library for running many small, specialized agents against a live conversation, where the hard part is not producing text — it is deciding that almost all of the time, nothing should be said. The framework's job is to make silence the default and to make what does get through trustworthy.

What it does: selects which agents are eligible this turn, runs them concurrently, validates everything they produce at one acceptance boundary, and merges the results into a single response.

What your host owns: everything the user sees or feels. The engine never renders a card, never executes a UI action, and never decides what is on screen. It tells you what an agent is permitted to say and hands you a validated result; presentation, timing and action execution are yours.

Runtime: 3.2.0 · Python: 3.11, 3.12, 3.13 · Dependencies: openai, pydantic v2, jinja2.

Install

pip install -e .

From a clone, for development — the suite needs no API key and makes no network calls:

git clone https://github.com/genriq/xubb-agents.git
cd xubb-agents
pip install -e ".[dev]"
pytest

Run it now (no API key)

Start here rather than with a model: this exercises the real engine — eligibility, concurrency, validation, the acceptance boundary, the merge — with a rule-based agent instead of an LLM-backed one. No key, no network, deterministic output.

It demonstrates the machinery, not the quality of a language model's judgement.

import asyncio
from xubb_agents import AgentEngine, AgentContext, Blackboard
from xubb_agents.core.agent import BaseAgent, AgentConfig
from xubb_agents.core.models import AgentResponse, TranscriptSegment

class HesitationCoach(BaseAgent):
    """No LLM: fires a fixed nudge when the customer sounds hesitant."""
    async def evaluate(self, context):
        last = context.recent_segments[-1].text.lower() if context.recent_segments else ""
        if any(word in last for word in ("not sure", "budget", "expensive", "too much")):
            return AgentResponse(insights=[
                self.create_insight("Acknowledge the budget concern, then ask what would make it a fit."),
            ])
        return None  # silent otherwise — silence is the default

async def main():
    engine = AgentEngine()  # no api_key: this agent makes no LLM call
    engine.register_agent(HesitationCoach(AgentConfig(name="Hesitation Coach")))
    context = AgentContext(
        session_id="demo",
        recent_segments=[TranscriptSegment(speaker="customer", timestamp=0.0,
                                            text="I'm not sure this fits our budget...")],
        blackboard=Blackboard(),
    )
    response = await engine.process_turn(context)
    for insight in response.insights:
        print(f"[{insight.type.value}] {insight.content}")

asyncio.run(main())

Writing a BaseAgent subclass like this is a first-class path, not a toy: custom producers go through exactly the same validation as LLM-backed ones.

With a model

The same shape, with a DynamicAgent — an agent defined by data rather than code. This one makes one real provider call, so it needs credentials you supply.

Two things to expect. output_format is set explicitly; omitting it inherits a deprecated format and warns. And the agent may say nothing at all — an empty insights list is a normal, successful turn, not a failure.

import asyncio, os
from xubb_agents import AgentEngine, DynamicAgent, AgentContext, Blackboard
from xubb_agents.core.models import TranscriptSegment

agent = DynamicAgent({
    "id": "echo-coach",
    "name": "Echo Coach",
    "text": "You observe a live conversation. If the customer sounds hesitant, "
            "give the salesperson ONE short, concrete suggestion.",
    "output_format": "insight_v1",          # always set this explicitly
    "trigger_config": {"mode": "turn_based", "cooldown": 0},
})

async def main():
    engine = AgentEngine(api_key=os.environ.get("OPENAI_API_KEY"))
    engine.register_agent(agent)
    context = AgentContext(
        session_id="demo",
        recent_segments=[
            TranscriptSegment(speaker="customer", timestamp=0.0,
                              text="I'm not sure this fits our budget..."),
            TranscriptSegment(speaker="sales", timestamp=4.2,
                              text="What range did you have in mind?"),
        ],
        blackboard=Blackboard(),
    )
    response = await engine.process_turn(context)

    print(f"acceptance: {response.acceptance_status}")
    for insight in response.insights:
        # insights are AgentInsight objects, not dicts — use attribute access
        print(f"[{insight.type.value}] {insight.content}")
    if not response.insights:
        print("(silent — nothing was worth saying this turn)")

asyncio.run(main())

Reading the result

A turn returns one AgentResponse. Three things on it answer most questions:

insights	What to show. Empty is valid and common.
acceptance_status	accepted, accepted_silent, or rejected. Rejection is whole-response — a rejected agent commits nothing at all, so a diagnostic never means "some of this applied". On the merged turn response this is an aggregate: one rejected agent does not fail the turn, so read acceptance_by_agent when you need to know which agent was refused.
diagnostics	Why something was refused. Codes are documented in DIAGNOSTICS.md.

On an insight, read type for its purpose, content for the text, and urgency for when it needs to be seen. One field deserves a warning: confidence is 1.0 on insights whose model offered no estimate, so it is not evidence that confidence was assessed — confidence_provided is the field that tells you.

Not every diagnostic is a failure: capability_unavailable and unsupported_structured_output appear on accepted responses. Read acceptance_status alongside them.

Where to go next

If you want to	Read
Understand how to use this well	Design guide — restraint, specialized observers, what the engine owns and what you do
Write an agent	Authoring agents
Coordinate several agents	Orchestration
Wire it into your application	Host integration
Generate long-form output	Long-form content
Look up a class or field	API reference
Look up a diagnostic code	Diagnostics
Move off a deprecated output format	Migration

Release history is in the changelog; the sections below are reference material you do not need to read front to back.

---

Usage in Other Projects

from xubb_agents import AgentEngine, DynamicAgent, AgentContext, Blackboard, TriggerType

# Initialize engine
engine = AgentEngine(api_key="your-openai-key")

# Load agent from config
agent = DynamicAgent(config_dict)
engine.register_agent(agent)

# Create session with Blackboard
blackboard = Blackboard()

# Process conversation
context = AgentContext(
    session_id="session_123",
    recent_segments=[...],
    blackboard=blackboard,
    turn_count=1
)
response = await engine.process_turn(context, trigger_type=TriggerType.TURN_BASED)

What's New

v3.0 — one insight contract (BREAKING). legacy_v2 is removed: the insight_contract constructor argument is gone (passing it raises TypeError rather than being silently ignored), descriptors no longer declare supported_contracts, and every schema must name a typed_adapter. custom1 is removed and is refused by name — deleting its file alone would have let the loader's missing-file fallback resolve such an agent to default, a different envelope, without saying so. There is no partial acceptance any more: a recoverable insight error rejects the whole response. The legacy adapter coercions (case-folding a type, defaulting an absent one to "suggestion") are gone with it — typed asks for an exact value and reports what the model wrote. See SPEC_REMOVE_LEGACY_CONTRACT and the CHANGELOG for the migration table.

3.1.0 — two output formats, and seven contract disagreements repaired. Authoring consolidates onto insight_v1 (insight + state) and widget_control (insight + state + validated UI actions), both on one canonical envelope: an explicit Boolean has_insight gate and a nested insight. default, default_v2, v2_raw and ui_control become thin input adapters over the same validation pipeline — deprecated here, removed in 4.0.0 — and every format's gate, keys and channels now come from one authority (library/contract/output_formats.json) that the generated prompt, the provider projection and the parser are all derived from. Repaired with it: default published message while the parser read content; a custom gate or root name changed the parser but not the prompt; speak_without_gate was accepted and inert; an undeclared channel was committed when an agent stayed silent and rejected when it spoke; a UI action could be any shape; an unknown format name silently became default; and the provider projection required state_updates and data of the model on a format whose parser read neither. UI actions are now a host capability — AgentContext.widget_capabilities declares the permitted targets, actions and payload keys, missing declarations authorize nothing, and the instruction the model gets is built from those declarations. See SPEC_OUTPUT_FORMAT_CONSOLIDATION and MIGRATION_OUTPUT_FORMATS.

v2.8 — typed reach. Every shipped schema but custom1 registers under the typed contract, and custom1 says why it cannot. (3.1.0 supersedes the adapter names in this paragraph: the three envelope shapes are now canonical, flat and root, and widget_control is canonical.) Typed cards carry evidence coordinates (EvidenceRef.source_index: the segment's position in the list the host passed, before trimming) and a host may ask for citation markers on every run (evidence_citations); the isolated content path's instruction is result-only; urgency_provided says whether the model authored the urgency; await handle.result() returns only after the content slot is released (handle.released); PriorInsightRecord.correctable controls what may be corrected; data_by_agent attributes sidecars per agent. Additive; the legacy contract is untouched. See SPEC_V2_8_TYPED_REACH. 2.8.1 adds one diagnostic detail: when no JSON object arrives, invalid_envelope carries the model client's failure category (timeout, refusal, malformed, …) instead of none.

v2.7 — the insight contract (XUBB-ITC-1 1.2.0) and the hardening audit. Nine human-facing insight purposes (fact/information, observation, suggestion, warning, opportunity, praise, reply, correction, question) behind an opt-in typed contract — AgentEngine(insight_contract="typed_v1") — with strict local validation, whole-response atomic rejection, engine-minted identity, evidence catalogs, permissioned replies and questions with a correlated answer channel, a correction lifecycle with arbitration, provider structured outputs derived from the shipped contract, the negotiated long_form_v1 content contract, and isolated active-session content tasks. The legacy contract stays the default and keeps its wire shape, but is safer: unknown types are rejected rather than relabelled, recoverable insight errors commit valid channels with an explicit partial status, and framework ERROR cards are sanitized. An independent audit then drove three hardening increments: one acceptance boundary for every producer (custom agents and callback-modified responses included), per-agent answer scoping, present-and-matching principal identity for interactive operations, typed failures as diagnostics, entrypoint admission and explicit lifecycle for content tasks, contract-derived prompts, strict content-policy primitives, and a clean-wheel CI job. The audit's own regression suite is adopted verbatim and passes. See the changelog for the migration notes.

v2.6 / v2.5 — modern-model compatibility. Any OpenAI Chat Completions model is a valid per-agent choice — gpt-4.x, gpt-5.x, o-series — with cheap-by-default economics: whisper agents stay fast and near-free, reasoning is an explicit per-agent opt-in (reasoning_effort, per-agent timeout/max_tokens, a validated model_params passthrough, and per-call token-usage telemetry). One deliberate edge: registering a reasoning-capable model without an explicit reasoning_effort fails at load time with a copy-pasteable fix (strict_reasoning_config=False downgrades it to a warning).

Full release notes in the CHANGELOG; design rationale in SPEC_LLM_MODERN_MODELS. Earlier releases — v2.2 (hardening), v2.1 (robustness), v2.0 (structured-blackboard rewrite) — are summarized there too, with archived specs under docs/archive/.

Architecture

┌─────────────────────────────────────────────────────────────────────────┐
│                         HOST APPLICATION                                 │
│  (web server, CLI tool, desktop app, etc.)                             │
│                                                                          │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐                   │
│  │ Transcription│  │ Preprocessing│  │   Session    │                   │
│  │   Source     │─▶│   (optional) │─▶│   Manager    │                   │
│  └──────────────┘  └──────────────┘  └──────┬───────┘                   │
│                                              │                           │
└──────────────────────────────────────────────┼───────────────────────────┘
                                               │
                                               ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                      XUBB_AGENTS FRAMEWORK                               │
│                                                                          │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │                       AgentContext                               │    │
│  │  - session_id          - recent_segments    - shared_state      │    │
│  │  - trigger_type        - trigger_metadata   - rag_docs          │    │
│  │  - user_context        - language_directive - blackboard (v2)   │    │
│  └─────────────────────────────────────────────────────────────────┘    │
│                                    │                                     │
│                                    ▼                                     │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │                       AgentEngine                                │    │
│  │                                                                  │    │
│  │  Phase 1: Run TURN_BASED/KEYWORD/SILENCE/INTERVAL agents        │    │
│  │      │                                                          │    │
│  │      ▼                                                          │    │
│  │  Collect Events + Apply State Updates                           │    │
│  │      │                                                          │    │
│  │      ▼                                                          │    │
│  │  Phase 2: Run EVENT-triggered agents (if events emitted)        │    │
│  │      │                                                          │    │
│  │      ▼                                                          │    │
│  │  Aggregate Responses                                            │    │
│  └─────────────────────────────────────────────────────────────────┘    │
│                                    │                                     │
│                                    ▼                                     │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │                      AgentResponse                               │    │
│  │  - insights            - events             - variable_updates  │    │
│  │  - queue_pushes        - facts              - memory_updates    │    │
│  │  - data                - debug_info   - memory_updates_by_agent │    │
│  └─────────────────────────────────────────────────────────────────┘    │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

Component Responsibilities

• AgentEngine: Orchestrates agent execution, manages Blackboard, handles multi-phase execution
• BaseAgent: Abstract base class defining agent interface
• DynamicAgent: Configurable agent loaded from prompts/database
• Blackboard: Structured shared state (Variables, Events, Queues, Facts, Memory)
• ConditionEvaluator: Evaluates trigger conditions against Blackboard state
• LLMClient: Isolated OpenAI client for agent LLM calls

To learn how to design and compose agents well, see the design guide and the task guides it links to. For detailed implementation and data models, see technical_spec_agents.md. For prompt writing best practices, see prompt_engineering_guide.md.

---

Core Concepts

1. Agent

An agent is a self-contained unit of intelligence that:

• Observes conversation context
• Evaluates whether to provide insight
• Returns structured insights (suggestions, warnings, facts)
• Updates Blackboard state for other agents
• Emits events to trigger other agents

2. Blackboard

The structured shared workspace where agents read and write information. Contains five containers:

Container	Purpose	Lifetime
Variables	Session-scoped key-value state	Session
Events	Transient signals to trigger other agents	Cleared after turn
Queues	Ordered work items (FIFO)	Session
Facts	Extracted knowledge with confidence	Session
Memory	Agent-private scratchpad	Session

3. Trigger

A trigger is an event that causes an agent to evaluate. Six types:

Type	Description	Use Case
TURN_BASED	After a speaker finishes a turn	General conversation analysis
KEYWORD	Keyword detected in transcript	Price objections, compliance alerts
SILENCE	Dead air threshold exceeded	Meeting facilitation
INTERVAL	Time-based periodic	Background monitoring
EVENT	Another agent emitted an event	Agent coordination
FORCE	User-triggered, bypasses cooldown and conditions	Manual intervention, debugging

4. Trigger Conditions

Preconditions that must be satisfied before an agent runs. Prevents unnecessary LLM calls.

{
  "trigger_conditions": {
    "mode": "all",
    "rules": [
      {"var": "phase", "op": "eq", "value": "negotiation"},
      {"fact": "budget", "op": "exists"}
    ]
  }
}

5. Insight

A piece of advice returned by an agent:

{
    "agent_id": "sales-coach",
    "agent_name": "Sales Coach",
    "type": "warning",  # suggestion, warning, opportunity, fact, praise, error
    "content": "Price objection detected. Focus on value.",
    "confidence": 0.9,
    "expiry": 15,  # seconds
    "action_label": "Handle Objection",  # Optional button text
    "metadata": {"zone": "A", "color": "red"}  # Optional UI hints (see zone note below)
}

Insight Types: (the "Zone" labels are an example UI taxonomy from the reference host — an urgency-based screen placement convention; hosts are free to ignore or remap them)

• SUGGESTION: Passive advice (Zone C)
• WARNING: Urgent negative alert (Zone A)
• OPPORTUNITY: Urgent positive alert (Zone A)
• FACT: Contextual information (Zone C)
• PRAISE: Positive reinforcement
• Interactive purposes, available once their supporting capability is implemented and the host declares it: OBSERVATION, REPLY, CORRECTION, QUESTION; INFORMATION is an alias of FACT (wire value stays "fact"). See docs/SPEC_INSIGHT_TYPES.md.
• ERROR: System issues

---

Blackboard Architecture

Container Overview

┌─────────────────────────────────────────────────────────────────────────┐
│                            BLACKBOARD                                    │
├─────────────────────────────────────────────────────────────────────────┤
│  EVENTS (transient, structured)                                          │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │ Event(name="question_detected", payload={"q": "What's price?"}) │    │
│  └─────────────────────────────────────────────────────────────────┘    │
├─────────────────────────────────────────────────────────────────────────┤
│  VARIABLES (session-scoped)                                              │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │ phase: "negotiation", sentiment: 0.7, turn_count: 15            │    │
│  └─────────────────────────────────────────────────────────────────┘    │
├─────────────────────────────────────────────────────────────────────────┤
│  QUEUES (ordered lists)                                                  │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │ pending_questions: ["What's the price?", "When can you start?"] │    │
│  └─────────────────────────────────────────────────────────────────┘    │
├─────────────────────────────────────────────────────────────────────────┤
│  FACTS (extracted knowledge)                                             │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │ {type: "budget", value: "$50,000", confidence: 0.9}             │    │
│  └─────────────────────────────────────────────────────────────────┘    │
├─────────────────────────────────────────────────────────────────────────┤
│  MEMORY (agent-private)                                                  │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │ question_extractor: {questions_found: 3, last_question: "..."}  │    │
│  └─────────────────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────────────────┘

Accessing Blackboard in Templates (Jinja2)

{# Variables #}
Current phase: {{ blackboard.variables.phase }}
Sentiment: {{ blackboard.variables.sentiment }}

{# Queues #}
Pending questions: {{ blackboard.queues.pending_questions | length }}
{% for q in blackboard.queues.pending_questions %}
- {{ q }}
{% endfor %}

{# Facts #}
{% for fact in blackboard.facts %}
- {{ fact.type }}: {{ fact.value }} (confidence: {{ fact.confidence }})
{% endfor %}

{# Memory (own) #}
My counter: {{ blackboard.memory[agent_id].counter }}

---

Trigger System

Turn-Based Triggers (Default)

Agents run after a speaker finishes a turn.

{
  "trigger_config": {
    "mode": "turn_based",
    "cooldown": 10
  }
}

Keyword-Based Triggers

Agents run immediately when a keyword is detected.

{
  "trigger_config": {
    "mode": "keyword",
    "keywords": ["price", "discount", "budget"],
    "cooldown": 5
  }
}

Note: Keyword detection is host responsibility. The engine provides check_keyword_triggers(text) as a helper.

Silence-Based Triggers

Agents run after a period of silence.

{
  "trigger_config": {
    "mode": "silence",
    "silence_threshold": 30,
    "cooldown": 10
  }
}

Interval-Based Triggers

Agents run on a periodic timer.

{
  "trigger_config": {
    "mode": "interval",
    "trigger_interval": 60,
    "cooldown": 5
  }
}

Event-Based Triggers (NEW in v2.0)

Agents run when another agent emits a specific event.

{
  "trigger_config": {
    "mode": "event",
    "subscribed_events": ["question_detected", "objection_raised"],
    "cooldown": 0
  }
}

Trigger Conditions

Define preconditions that must be satisfied:

{
  "trigger_conditions": {
    "mode": "all",
    "rules": [
      {"var": "phase", "op": "in", "value": ["negotiation", "closing"]},
      {"queue": "pending_questions", "op": "not_empty"},
      {"meta": "turn_count", "op": "gte", "value": 3}
    ]
  }
}

Available Operators: eq, neq, gt, gte, lt, lte, in, not_in, contains, exists, present, not_exists, not_empty, empty, mod

---

Agent Configuration

Full Configuration Schema

{
  "id": "string (unique identifier)",
  "name": "string (display name)",
  
  "trigger_config": {
    "mode": "turn_based | keyword | silence | interval | event | [array]",
    "cooldown": 15,
    "keywords": ["price", "discount"],
    "silence_threshold": 5,
    "subscribed_events": ["question_detected"]
  },
  
  "trigger_conditions": {
    "mode": "all | any",
    "rules": [
      {"var": "phase", "op": "eq", "value": "negotiation"}
    ]
  },
  
  "priority": 0,
  
  "model_config": {
    "model": "gpt-4o-mini",
    "context_turns": 30
  },
  
  "text": "Your system prompt here with {{ jinja2 }} templating",

  "output_format": "insight_v1 | widget_control",

  "include_context": true
}

Configuration Fields

Field	Type	Default	Description
id	string	auto-generated	Unique identifier
name	string	required	Display name
trigger_config.mode	string/array	"turn_based"	When to trigger
trigger_config.cooldown	int	15	Seconds between runs
trigger_config.keywords	array	[]	Keywords for KEYWORD trigger
trigger_config.silence_threshold	int	null	Seconds for SILENCE trigger
trigger_config.subscribed_events	array	[]	Events for EVENT trigger
trigger_conditions	object	null	Preconditions (Blackboard-aware)
priority	int	0	State update priority (higher wins)
model_config.model	string	"gpt-4o-mini"	LLM model (xubb_agents.DEFAULT_MODEL)
model_config.context_turns	int	6	Transcript segments to include
model_config.reasoning_effort	string	null	v2.6: explicit reasoning effort ("none"/"minimal"/"low"/…). Sent only when set — never injected. Required (hard-fail at registration) when the model name looks reasoning-capable; see strict_reasoning_config.
model_config.timeout	float	null	v2.6: per-agent request timeout (falls back to the client budget). Deep-effort agents need > 10s.
model_config.max_tokens	int	null	v2.6: per-agent token cap (wire: max_completion_tokens; includes reasoning tokens — deep-effort agents need ≥ 4096, OpenAI suggests ~25000).

---

Agent Communication

Event-Driven Coordination

Agent A (Emitter):

{
  "has_insight": false,
  "events": [
    {"name": "question_detected", "payload": {"question": "What is pricing?", "speaker": "CUSTOMER"}}
  ]
}

Agent B (Subscriber):

{
  "trigger_config": {
    "mode": "event",
    "subscribed_events": ["question_detected"]
  }
}

Multi-Phase Execution

1. Phase 1: TURN_BASED, KEYWORD, SILENCE, INTERVAL agents run in parallel
2. State Update: Variables, queues, facts, memory merged (priority-ordered)
3. Event Collection: Events gathered from Phase 1 responses
4. Phase 2: EVENT-triggered agents run (if events were emitted)
5. Cleanup: Events cleared from Blackboard

Priority System

Higher-priority agents can override lower-priority state updates:

# Agent A (priority=5): variable_updates={"phase": "closing"}
# Agent B (priority=10): variable_updates={"phase": "negotiation"}

# Result: phase = "negotiation" (Agent B wins, higher priority)

Updates are applied in ascending priority order (low → high), so higher priority writes last and wins.

Facts use the same priority precedence (v2.2 — F-1): when two agents emit facts that collide on (type, key), the winner is decided by priority first, then confidence, then registration order. The engine stamps each fact's priority from the emitting agent at merge time.

Private Memory

Each agent maintains a private scratchpad via memory_updates:

{
  "memory_updates": {
    "last_objection": "price",
    "objection_count": 4
  }
}

Access in templates: {{ blackboard.memory[agent_id].objection_count }}

Memory persists across turns via the blackboard. The engine syncs each agent's committed memory into shared_state["memory_<agent_id>"] before agents run (v2.2 — MR-1), so persistent memory survives even when the host re-instantiates agents every turn. Memory values are deep-copied on both read and write (v2.2 — M-1), so a caller mutating an object it passed into update_memory never mutates blackboard state.

---

Usage Guide

Basic Integration

from xubb_agents import (
    AgentEngine, AgentContext, TriggerType, Blackboard,
    DynamicAgent, TranscriptSegment
)

# Initialize
engine = AgentEngine(api_key="sk-...")

# Load agents from config
for config in load_agents_from_db():
    engine.register_agent(DynamicAgent(config))

# Session state
session = {
    "id": "session_123",
    "blackboard": Blackboard(),
    "turn_count": 0,
    "segments": []
}

async def on_transcript_segment(segment: dict):
    """Called when speech is detected."""
    session["segments"].append(TranscriptSegment(**segment))
    session["turn_count"] += 1
    
    context = AgentContext(
        session_id=session["id"],
        recent_segments=session["segments"][-100:],
        blackboard=session["blackboard"],
        turn_count=session["turn_count"],
        user_context="Sales Director at Acme Corp"
    )
    
    response = await engine.process_turn(
        context,
        trigger_type=TriggerType.TURN_BASED
    )
    
    for insight in response.insights:
        await send_to_ui(insight)

Keyword Handling

async def on_keyword_detected(keyword: str, segment: dict):
    """Called when a keyword is detected."""
    matches = engine.check_keyword_triggers(segment["text"])
    allowed_ids = [agent.config.id for agent, _ in matches]
    
    context = AgentContext(
        session_id=session["id"],
        recent_segments=session["segments"][-100:],
        blackboard=session["blackboard"],
        trigger_metadata={"keyword": keyword}
    )
    
    response = await engine.process_turn(
        context,
        allowed_agent_ids=allowed_ids,
        trigger_type=TriggerType.KEYWORD
    )

Creating a Custom Agent

from xubb_agents.core.agent import BaseAgent, AgentConfig
from xubb_agents.core.models import AgentContext, AgentResponse, InsightType, TriggerType, Event

class MyAgent(BaseAgent):
    def __init__(self):
        super().__init__(AgentConfig(
            name="My Agent",
            id="my-agent",
            cooldown=10,
            trigger_types=[TriggerType.TURN_BASED],
            priority=5
        ))
    
    async def evaluate(self, context: AgentContext) -> AgentResponse:
        response = AgentResponse()
        
        last_text = context.recent_segments[-1].text
        
        if "urgent" in last_text.lower():
            response.insights.append(
                self.create_insight(
                    content="Urgency detected. Escalate if needed.",
                    type=InsightType.WARNING
                )
            )
            # Emit event for other agents
            response.events.append(Event(
                name="urgency_detected",
                payload={"text": last_text},
                source_agent=self.config.id
            ))
        
        return response

Enabling Debugging

from xubb_agents.utils.tracing import StructuredLogTracer
from xubb_agents.core.callbacks import AgentCallbackHandler

tracer = StructuredLogTracer()
engine = AgentEngine(api_key="...", callbacks=[tracer])

# Traces include multi-phase execution details

Writing Custom Callbacks

Subclass AgentCallbackHandler and override only the methods you need — all are no-op by default:

class MyCallback(AgentCallbackHandler):
    async def on_turn_start(self, context: AgentContext) -> None: ...
    async def on_turn_end(self, response: AgentResponse, duration: float) -> None: ...
    async def on_agent_start(self, agent_name: str, context: AgentContext) -> None: ...
    async def on_agent_finish(self, agent_name: str, response: Optional[AgentResponse],
                              duration: float) -> None: ...
    async def on_agent_error(self, agent_name: str, error: Exception) -> None: ...
    async def on_agent_skipped(self, agent_name: str, reason: str) -> None: ...
    async def on_phase_start(self, phase: int, agent_names: List[str]) -> None: ...
    async def on_phase_end(self, phase: int, event_names: List[str]) -> None: ...
    async def on_chain_error(self, error: Exception) -> None: ...

Callback failures are non-fatal — they are logged and never abort turn processing.

---

API Reference

The full reference now lives in docs/API_REFERENCE.md — every supported class, method, parameter and field, with the facts generated from the shipped code and the meaning written by hand. Diagnostic codes are in docs/DIAGNOSTICS.md. Both are gated by tools/check_api_docs.py, so they cannot drift from the code.

The summary below is a quick orientation, not the reference.

Historic note: this section was incomplete and is being rebuilt. A 2026-09 audit found it covers roughly half the public surface — it predates the insight contract (2.7), typed reach (2.8) and the content contract, and omits AgentInsight's typed fields, AgentResponse's acceptance fields and most engine parameters. Until the rebuild lands, treat the docstrings in src/xubb_agents/ as authoritative where the two disagree, and see MIGRATION_OUTPUT_FORMATS.md for the current output-format and widget-capability surface.

AgentEngine

class AgentEngine:
    def __init__(self, api_key: Optional[str] = None,
                 callbacks: List[AgentCallbackHandler] = None,
                 max_phases: int = 2, **engine_options)
        # api_key is OPTIONAL: an engine can be constructed without one (the
        # LLM client is created lazily). `engine_options` stands in for ten
        # further parameters this reference does not yet list — see the note
        # above and, until it is rebuilt, the constructor docstring in
        # src/xubb_agents/core/engine.py.
    def register_agent(self, agent: BaseAgent) -> None
    def replace_agents(self, agents: List[BaseAgent]) -> None
        # Atomic full-registry swap for hot reloads: rebuilds the registry and
        # rebinds it LAST, so a concurrent turn never observes a half-cleared
        # registry. Prefer this over clear()+register loops.
        # (update_api_key remains non-concurrency-safe.)
    def update_api_key(self, api_key: str) -> None
    async def process_turn(
        self,
        context: AgentContext,
        allowed_agent_ids: Optional[List[str]] = None,
        trigger_type: TriggerType = TriggerType.TURN_BASED,
        trigger_metadata: Dict[str, Any] = None
    ) -> AgentResponse
    def check_keyword_triggers(
        self,
        text: str,
        allowed_agent_ids: Optional[List[str]] = None
    ) -> List[tuple]

Blackboard

class Blackboard(BaseModel):
    events: List[Event]
    variables: Dict[str, Any]
    queues: Dict[str, List[Any]]
    facts: List[Fact]
    memory: Dict[str, Dict[str, Any]]
    
    # Event operations
    def emit_event(self, event: Event) -> None
    def has_event(self, event_name: str) -> bool
    def count_events(self, event_name: str) -> int
    def get_events_by_name(self, event_name: str) -> List[Event]
    def clear_events(self) -> None

    # Variable operations (sys.* keys are engine-reserved)
    def set_var(self, key: str, value: Any) -> None
    def get_var(self, key: str, default: Any = None) -> Any
    def has_var(self, key: str) -> bool
    def delete_var(self, key: str) -> None

    # Queue operations
    def push_queue(self, queue_name: str, item: Any) -> None
    def push_queue_items(self, queue_name: str, items: List[Any]) -> None
    def pop_queue(self, queue_name: str) -> Optional[Any]
    def peek_queue(self, queue_name: str) -> Optional[Any]
    def queue_length(self, queue_name: str) -> int
    def has_queue(self, queue_name: str) -> bool
    def clear_queue(self, queue_name: str) -> None

    # Fact operations
    def add_fact(self, fact: Fact) -> None
    def get_fact(self, fact_type: str, key: Optional[str] = None) -> Optional[Fact]
    def get_facts_by_type(self, fact_type: str) -> List[Fact]
    def has_fact(self, fact_type: str, key: Optional[str] = None) -> bool

    # Memory operations (get_memory returns a deep copy)
    def get_memory(self, agent_id: str) -> Dict[str, Any]
    def set_memory(self, agent_id: str, data: Dict[str, Any]) -> None
    def update_memory(self, agent_id: str, updates: Dict[str, Any]) -> None
    def has_memory(self, agent_id: str) -> bool

    # Serialization
    def snapshot(self) -> Blackboard
    def to_dict(self) -> Dict[str, Any]
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> Blackboard

AgentContext

class AgentContext(BaseModel):
    session_id: str
    recent_segments: List[TranscriptSegment]
    
    # State (v1 compatibility)
    shared_state: Dict[str, Any] = {}
    
    # Blackboard (v2)
    blackboard: Optional[Blackboard] = None
    
    # Trigger information
    trigger_type: TriggerType = TriggerType.TURN_BASED
    trigger_metadata: Dict[str, Any] = {}
    
    # Context enrichment
    rag_docs: List[str] = []
    user_context: Optional[str] = None
    language_directive: Optional[str] = None
    
    # Execution metadata
    turn_count: int = 0
    phase: int = 1

    # Role overrides (per-agent config modifications)
    agent_config_overrides: Dict[str, AgentConfigOverride] = {}

AgentConfigOverride

class AgentConfigOverride(BaseModel):
    cooldown_modifier: Optional[int] = None       # +N = slower, -N = faster (floor 5s)
    context_turns_modifier: Optional[int] = None   # +N = more context, -N = less
    instructions_append: Optional[str] = None      # Extra instructions appended to prompt

AgentResponse

class AgentResponse(BaseModel):
    # Agent identity (v2.1 — set by framework)
    source_agent_id: Optional[str] = None

    # Core output
    insights: List[AgentInsight] = Field(default_factory=list)

    # Blackboard updates (v2)
    events: List[Event] = Field(default_factory=list)
    variable_updates: Dict[str, Any] = Field(default_factory=dict)
    queue_pushes: Dict[str, List[Any]] = Field(default_factory=dict)
    facts: List[Fact] = Field(default_factory=list)
    memory_updates: Dict[str, Any] = Field(default_factory=dict)

    # Per-agent keyed memory (v2.1.1 — populated on aggregated responses from process_turn)
    memory_updates_by_agent: Dict[str, Dict[str, Any]] = Field(default_factory=dict)

    # Legacy compatibility (v1)
    state_updates: Dict[str, Any] = Field(default_factory=dict)

    # Sidecar data
    data: Dict[str, Any] = Field(default_factory=dict)
    debug_info: Dict[str, Any] = Field(default_factory=dict)

AgentConfig

class AgentConfig:
    def __init__(self,
        name: str,                                          # Display name (required)
        id: str = None,                                     # Unique ID (auto-generated from name)
        cooldown: int = 10,                                 # Seconds between runs
        model: str = "gpt-4o-mini",                         # LLM model
        trigger_types: List[TriggerType] = [TriggerType.TURN_BASED],
        trigger_keywords: List[str] = [],                   # For KEYWORD trigger
        silence_threshold: Optional[int] = None,            # For SILENCE trigger
        trigger_interval: Optional[int] = None,             # For INTERVAL trigger
        priority: int = 0,                                  # Merge priority (higher wins)
        output_format: str = "default",                     # deprecated default; set it explicitly.
                                                            # 4.0.0 (approved, unshipped): required.
        trigger_conditions: Optional[Dict] = None,          # Blackboard preconditions
        subscribed_events: Optional[List[str]] = None,      # For EVENT trigger
        # v2.6 — per-agent LLM-call config (declarative on custom subclasses:
        # the subclass owns forwarding these into its own generate() calls)
        reasoning_effort: Optional[str] = None,             # Sent only when set (INV-15)
        timeout: Optional[float] = None,                    # Per-agent request timeout
        max_tokens: Optional[int] = None,                   # Per-agent token cap
        model_params: Optional[Dict[str, Any]] = None       # Wire passthrough (validated)
    )

Data Models

class Event(BaseModel):
    name: str
    payload: Dict[str, Any] = {}
    source_agent: str
    timestamp: float
    id: Optional[str] = None  # Optional unique ID for tracing/deduplication

class Fact(BaseModel):
    type: str
    key: Optional[str] = None
    value: Any
    confidence: float = 1.0
    priority: int = 0          # v2.2 — engine-stamped emitting-agent priority (conflict resolution)
    source_agent: str
    timestamp: float           # session-relative seconds

Fact conflict resolution (v2.2): Facts colliding on (type, key) resolve by priority first, then confidence, then registration order. priority is stamped by the engine at merge time from the emitting agent's priority — agents should not set it; hosts calling add_fact() directly own it.

class AgentInsight(BaseModel):
    agent_id: str
    agent_name: str
    type: InsightType
    content: str
    confidence: float = 1.0
    expiry: int = 15
    action_label: Optional[str] = None
    metadata: Dict[str, Any] = {}

---

Best Practices

1. Use Trigger Conditions: Prevent unnecessary LLM calls with preconditions
2. Emit Events: Use events for agent coordination instead of polling state
3. Set Appropriate Cooldowns: 10-30 seconds typical to prevent spam
4. Use Priority Wisely: Higher priority for critical agents (compliance, escalation)
5. Limit Context Window: 6-10 turns for cost efficiency
6. Use Facts for Knowledge: Structured extraction with confidence scores
7. Use Queues for Work Items: FIFO processing of pending tasks

---

Migration from v1.0 / v2.0

v2.2 behavioral change (F-1): Facts that collide on (type, key) now resolve by priority first (then confidence, then registration order), matching the always-documented contract. Consumers that depended on the prior confidence-only behavior may see a different fact win. Fact.priority is a new defaulted field — serialized v2.1.1 facts load unchanged with priority=0. All other v2.2 changes are additive or internal. See SPEC_V2_2_HARDENING.md §13 (Migration Notes).

v2.0/v2.1 remain backward compatible; existing agents work unchanged. The v2.1 behavioral normalizations (callbacks fire once, get_memory() returns a copy, Jinja2 sandboxed, cooldown after errors) are detailed in the CHANGELOG and archived specs. The v1.0 → v2 mapping is auto-applied:

v1.0 Pattern	v2.0 Equivalent	Auto-Mapped?
shared_state["key"]	blackboard.variables["key"]	✅ Yes
state_updates	variable_updates	✅ Yes
memory_{agent_id}	blackboard.memory[agent_id]	✅ Yes
{{ state.key }}	{{ blackboard.variables.key }}	✅ Both work

Gradual Migration:

1. Add trigger_conditions to reduce LLM calls
2. Add events for agent coordination
3. Use facts for extracted knowledge
4. Use queues for work items

---

License

MIT — see LICENSE.

Contributing

Issues and PRs are welcome. This framework is spec-driven: a behavioral change ships with its contract-registry entry and a rule-asserting test, and the contract gate (tools/check_contracts.py, enforced in CI) must stay green. See CONTRIBUTING.md and docs/PROCESS.md for the full workflow. Security issues: see SECURITY.md, not a public issue.