basion-agent 0.9.0
Python SDK for Basion AI Agent framework - handles agent registration, message consumption, and streaming responses
License: MIT
Requires Python: >=3.10

# Basion Agent SDK

Python SDK for building AI agents on the Basion platform. Agents register themselves, receive messages through Kafka, and stream responses back to users in real time.

## Table of Contents

- [Installation](#installation)
- [Quick Start](#quick-start)
- [CLI](#cli)
- [Configuration](#configuration)
- [Agent Registration](#agent-registration)
- [Message Handling](#message-handling)
  - [Sender Filtering](#sender-filtering)
  - [Designing Handlers for Agents vs Users](#designing-handlers-for-agents-vs-users)
- [Message](#message)
- [Streaming Responses](#streaming-responses)
  - [Streamer Options](#streamer-options)
  - [When to Use `awaiting=True`](#when-to-use-awaitingtrue)
  - [Response Schema (genui Forms)](#response-schema-genui-forms)
- [Conversation History](#conversation-history)
- [Memory V2 (mem0)](#memory-v2-mem0)
- [Attachments](#attachments)
- [Structural Streaming](#structural-streaming)
  - [Artifact](#artifact) · [Surface](#surface) · [TextBlock](#textblock) · [Stepper](#stepper)
- [Generative UI Message Components](#generative-ui-message-components)
- [Knowledge Graph](#knowledge-graph)
- [Agent Inventory](#agent-inventory)
- [Agent-Initiated (Proactive) Conversations](#agent-initiated-proactive-conversations)
- [Inter-Agent Communication](#inter-agent-communication)
  - [Hand-Off](#hand-off-messagehand_off) · [Call](#call-agentcall)
  - [Combining Patterns](#combining-patterns)
  - [When to Use Which](#when-to-use-which)
- [Remote Logging (Loki)](#remote-logging-loki)
- [Extensions](#extensions)
  - [LangGraph](#langgraph) · [Pydantic AI](#pydantic-ai)
- [Full Example](#full-example-rare-disease-assistant)
- [Architecture](#architecture)

## Installation

```bash
pip install basion-agent

# With LangGraph support
pip install basion-agent[langgraph]

# With Pydantic AI support
pip install basion-agent[pydantic]
```

## Quick Start

```python
from basion_agent import BasionAgentApp

app = BasionAgentApp(
    gateway_url="agent-gateway:8080",
    api_key="your-api-key",
)

agent = app.register_me(
    name="rare-disease-assistant",
    about="Answers questions about rare diseases, symptoms, and treatments",
    document="Rare disease specialist. Abilities: disease lookup, symptom analysis, genetic variant interpretation, finding similar conditions. Keywords: rare disease, genetics, orphan disease, diagnosis.",
)

@agent.on_message
async def handle(message, sender):
    history = await message.conversation.get_history(limit=10)

    async with agent.streamer(message) as s:
        s.stream(f"You asked: {message.content}\n\n")
        s.stream("Let me look into that for you...")

app.run()
```

That's all you need. When you call `app.run()`, the SDK registers your agent with the gateway, subscribes to `{agent-name}.inbox` on Kafka, starts a heartbeat loop, and begins consuming messages. Your `@on_message` handler fires for each incoming message.

## CLI

```bash
# Run your agent
basion-agent run main:app

# Defaults to :app
basion-agent run main

# Custom app name
basion-agent run myagent:application

# Show version
basion-agent version
```

## Configuration

```python
app = BasionAgentApp(
    gateway_url="agent-gateway:8080",       # Required
    api_key="key",                           # Required
    heartbeat_interval=60,                   # Heartbeat frequency in seconds
    max_concurrent_tasks=100,                # Max concurrent message handlers
    error_message_template="...",            # Error message sent to users on failure
    secure=False,                            # TLS for gRPC and HTTPS for HTTP
    enable_remote_logging=False,             # Send logs to Loki via gateway
    remote_log_level=logging.INFO,           # Min log level for remote logging
    remote_log_batch_size=100,               # Logs per batch
    remote_log_flush_interval=5.0,           # Seconds between flushes
)
```

| Environment Variable | Description |
|---|---|
| `GATEWAY_URL` | Agent Gateway endpoint |
| `GATEWAY_API_KEY` | API key for authentication |

## Agent Registration

```python
agent = app.register_me(
    name="rare-disease-assistant",            # Unique identifier (kebab-case, used for Kafka routing)
    about="Rare disease medical assistant",   # Short description for agent selection (max 500 chars)
    document="""Rare disease specialist. Abilities: disease lookup, symptom analysis,
    genetic variant interpretation, finding similar conditions, drug info.
    Keywords: rare disease, genetics, orphan disease, diagnosis, phenotype, genotype.""",
    representation_name="Dr. Assistant",      # Display name (defaults to name)
    version="1.0.0",                          # Version string
    lifecycle_stage="public",                 # 'dev' | 'test' | 'private-preview' | 'public-preview' | 'public'
    welcome_message="Hello! I specialize in rare diseases. How can I help?",
    detailed_description="A comprehensive rare disease assistant...",
    base_url="http://...",                    # Base URL for frontend service (optional)
    metadata={"specialty": "rare-diseases"},  # Free-form metadata (optional)
    category_names=["my-body", "medical"],    # Categories in kebab-case (auto-created)
    prompts=[                                 # Suggested prompts shown to users
        {"label": "Huntington", "prompt": "What is Huntington disease?"},
        {"label": "Similar diseases", "prompt": "Find diseases similar to Cystic Fibrosis"},
    ],
    waitlist=False,                           # Not publicly available yet (optional)
    related_pages=[                           # Related pages (optional)
        {"name": "Resources", "endpoint": "/resources"}
    ],
    force_update=False,                       # Bypass content hash check (optional)
)
```

| Parameter | Type | Default | Description |
|---|---|---|---|
| `name` | `str` | required | Unique agent name in kebab-case (used for Kafka routing) |
| `about` | `str` | required | Short description (max 500 characters, used for agent selection) |
| `document` | `str` | required | Used by the router to match user messages to your agent. Include keywords describing abilities and key capabilities so the router can properly route to your agent |
| `representation_name` | `str` | `name` | Human-readable display name |
| `version` | `str` | `None` | Version string (e.g., "1.0.0"). Included in content hash — changing version triggers re-registration |
| `lifecycle_stage` | `str` | `'dev'` | One of: `dev`, `test`, `private-preview`, `public-preview`, `public` |
| `welcome_message` | `str` | `None` | Welcome text shown when a user starts a conversation |
| `detailed_description` | `str` | `None` | Extended description beyond `about` |
| `base_url` | `str` | `None` | Base URL for agent's frontend service (used for iframe artifact URLs) |
| `metadata` | `dict` | `None` | Free-form metadata dict |
| `category_names` | `list[str]` | `None` | Categories in kebab-case (auto-created if they don't exist) |
| `prompts` | `list[dict]` | `None` | List of `{"label": "...", "prompt": "..."}` objects shown to users |
| `waitlist` | `bool` | `None` | If `True`, agent is not publicly available |
| `related_pages` | `list[dict]` | `None` | Pages with `name` and `endpoint` keys (supports `{conversation_id}` placeholder) |
| `force_update` | `bool` | `None` | Bypass content hash check and always update |

## Message Handling

### Basic Handler

```python
@agent.on_message
async def handle(message, sender):
    async with agent.streamer(message) as s:
        s.stream("Hello! How can I help?")
```

Handlers can be `async` or synchronous. Sync handlers run in a thread pool automatically.

### Sender Filtering

```python
# Handle messages from users only
@agent.on_message(senders=["user"])
async def handle_user(message, sender):
    ...

# Handle messages from a specific agent
@agent.on_message(senders=["triage-agent"])
async def handle_triage(message, sender):
    ...

# Exclude a specific sender
@agent.on_message(senders=["~notification-agent"])
async def handle_others(message, sender):
    ...
```

#### Designing Handlers for Agents vs Users

User messages and agent messages often require fundamentally different handling. A user message goes through your full conversation flow — history loading, memory, LLM calls. But when another agent calls you via `agent.call()`, it expects a focused, structured answer — not a full conversation experience.

**Scenario: A pharmacology agent with both user and agent interfaces.**

```python
# User-facing handler — full conversational flow with history, memory, streaming
@pharma_agent.on_message(senders=["user"])
async def handle_user(message, sender):
    history = await message.conversation.get_agent_history(limit=20)
    memories = await message.memory_v2.search(message.content)

    async with pharma_agent.streamer(message) as s:
        s.stream(generate_conversational_response(message.content, history, memories))

# Agent-facing handler — concise, structured response for agent.call()
@pharma_agent.on_message(senders=["coordinator-agent", "wellness-agent"])
async def handle_agent_call(message, sender):
    # No need for history/memory — the calling agent provides all context
    # Return a focused, structured answer
    async with pharma_agent.streamer(message) as s:
        s.stream(lookup_drug_interactions(message.content))
```

> **Tip:** When an agent calls yours via `agent.call()`, the response is automatically routed back to the caller — no special code needed. But if you have a dedicated `senders` handler, you can tailor the response format to what the calling agent expects (e.g., structured data vs conversational prose).

### Error Handling

If a handler throws, the SDK automatically sends an error message to the user. Customize or disable this:

```python
# Custom error message
app = BasionAgentApp(
    ...,
    error_message_template="Sorry, something went wrong. Please try again.",
)

# Or disable per-agent
agent.send_error_responses = False

@agent.on_message
async def handle(message, sender):
    try:
        ...
    except Exception as e:
        async with agent.streamer(message) as s:
            s.stream(f"I ran into an issue: {e}")
```

## Message

The `message` object provides access to content, conversation history, memory, and attachments.

```python
@agent.on_message
async def handle(message, sender):
    message.content              # Message text
    message.conversation_id      # Conversation UUID
    message.user_id              # User UUID
    message.metadata             # Optional dict (from frontend)
    message.schema               # Optional dict (for form responses)
    message.headers              # Raw Kafka headers
    message.conversation         # Conversation helper
    message.memory_v2            # mem0 memory (ingest + search)
    message.memory               # Deprecated memory (use memory_v2)

    # Inter-agent communication
    message.hand_off("agent-name")              # Forward to another agent
    message.hand_off("agent-name", "content")   # Forward with custom content
```

## Streaming Responses

### Basic Streaming

```python
@agent.on_message
async def handle(message, sender):
    # Context manager — auto-finishes on exit
    async with agent.streamer(message) as s:
        s.stream("Looking up information on Huntington disease...\n\n")
        s.stream("Huntington disease is a progressive neurodegenerative disorder...")

    # Or manual control
    s = agent.streamer(message)
    s.stream("Hello!")
    await s.finish()
```

`s.write(content)` is available as an alias for `s.stream(content)`.

### Streamer Options

```python
# Set awaiting — next user reply routes back to this agent
async with agent.streamer(message, awaiting=True) as s:
    s.stream("What symptoms are you experiencing?")
```

### When to Use `awaiting=True`

Use `awaiting=True` when you want to keep the user talking to your agent and **bypass the router** for the next message. This is powerful but requires care — you need to release the conversation at some point, or the user gets stuck with your agent forever.

**Scenario: Multi-turn data collection.** A diagnostic agent asks the user a series of questions. Each reply must come back to the same agent, not get rerouted by the router to a different specialist.

```python
@diagnostic_agent.on_message
async def handle(message, sender):
    history = await message.conversation.get_agent_history(limit=10)

    if len(history) == 0:
        # First interaction — ask the first question
        async with diagnostic_agent.streamer(message, awaiting=True) as s:
            s.stream("I'll help assess your symptoms. How long have you been experiencing this?")
        return

    if len(history) < 4:
        # Still collecting — keep awaiting
        async with diagnostic_agent.streamer(message, awaiting=True) as s:
            s.stream(ask_next_question(history, message.content))
        return

    # Collection done — respond WITHOUT awaiting to release back to router
    async with diagnostic_agent.streamer(message) as s:
        s.stream(generate_assessment(history, message.content))
```

**Scenario: Awaiting + hand-off for off-topic messages.** You hold the conversation with `awaiting=True`, but if the user asks something unrelated, you hand off to the guide agent instead of trying to answer it yourself.

```python
@nutrition_agent.on_message
async def handle(message, sender):
    if not is_nutrition_related(message.content):
        # Not our topic — hand off to the guide agent who can reroute properly
        message.hand_off("guide-agent", message.content)
        return

    # On-topic — keep the conversation
    async with nutrition_agent.streamer(message, awaiting=True) as s:
        s.stream(generate_nutrition_advice(message.content))
```

> **Tip:** Always have an exit path. If you set `awaiting=True` indefinitely without a way out, the user cannot reach other agents. Common exits: (1) stop setting `awaiting` after your task is done, (2) hand off to `guide-agent` when the topic doesn't match yours, or (3) use the agent inventory to find a better agent and hand off.

### Non-Persisted Content

Chunks that appear in real-time but don't get saved to conversation history:

```python
async with agent.streamer(message) as s:
    s.stream("Searching knowledge graph...", persist=False, event_type="thinking")
    # ... do work ...
    s.stream("Here are the results:")  # This gets persisted
```

### Message Metadata

Attach metadata to the response message:

```python
async with agent.streamer(message) as s:
    s.set_message_metadata({"source": "knowledge_graph", "confidence": 0.92})
    s.stream("Based on the knowledge graph, ...")
```

### Response Schema (genui Forms)

Request structured input from users with `genui` — a declarative form builder. Define form classes with typed fields, send them to the frontend, and parse typed responses back.

#### Quick Start

```python
from basion_agent.genui import Form, Text, Slider, MultiSelect, Switch, Option

class SymptomReport(Form):
    name = Text(label="Full Name", placeholder="John Doe")
    severity = Slider(label="Pain Level", min=1, max=10)
    symptoms = MultiSelect(label="Symptoms", options=[
        Option(value="headache", label="Headache"),
        Option(value="fatigue", label="Fatigue"),
        Option(value="nausea", label="Nausea"),
    ])
    chronic = Switch(label="Is this chronic?")
```

**Send to frontend:**

```python
@agent.on_message
async def handle(message, sender):
    # Check if this is a form submission
    if message.schema:
        data = await SymptomReport.parse(message)
        async with agent.streamer(message) as s:
            s.stream(f"Thank you, {data.name}. Pain level: {data.severity}/10.")
        return

    # Ask for structured input
    async with agent.streamer(message, awaiting=True) as s:
        s.stream("Please describe your symptoms:")
        form = SymptomReport(title="Symptom Report", description="Tell us about your symptoms")
        s.set_response_schema(form)
```

**How it works:**

1. Declare a `Form` subclass with field class attributes
2. Instantiate with metadata: `form = MyForm(title="...", description="...")`
3. `streamer.set_response_schema(form)` sends the schema to the frontend
4. When the user submits, `await MyForm.parse(message)` returns a typed Python object with IDE autocomplete

#### Field Types

All fields accept `required=True` (default). All arguments are keyword-only.

| Field | Python Type | None Default | Description |
|---|---|---|---|
| `Text` | `str` | `""` | Single/multiline text input |
| `Number` | `float` | `0.0` | Numeric input with min/max/step |
| `Select` | `str` | `""` | Single-select dropdown |
| `MultiSelect` | `List[str]` | `[]` | Multiple selection |
| `Checkbox` | `bool` | `False` | Single boolean checkbox |
| `CheckboxGroup` | `List[str]` | `[]` | Multiple checkboxes as a group |
| `Switch` | `bool` | `False` | Toggle switch |
| `Slider` | `float` | `0.0` | Range slider |
| `DatePicker` | `str` | `""` | Date selection |
| `Hidden` | `Any` | `None` | Hidden field (not rendered) |
| `FileField` | `AttachmentInfo` | `None` | File upload field |

**Text:**
```python
name = Text(label="Full Name", placeholder="John Doe", min_length=2, max_length=100)
bio = Text(label="Biography", multiline=True, required=False)
```

**Number:**
```python
age = Number(label="Age", min=0, max=150, step=1)
```

**Select / MultiSelect:**
```python
color = Select(label="Color", options=[
    Option(value="red", label="Red"),
    Option(value="blue", label="Blue"),
], placeholder="Pick one...", allow_custom=True)

symptoms = MultiSelect(label="Symptoms", options=[
    Option(value="headache", label="Headache"),
    Option(value="fatigue", label="Fatigue"),
], min_selections=1, max_selections=5, allow_custom=True)
```

**Checkbox / CheckboxGroup / Switch:**
```python
agree = Checkbox(label="I agree to the terms")
languages = CheckboxGroup(label="Languages", options=[
    Option(value="en", label="English"),
    Option(value="es", label="Spanish"),
], min_selections=1)
notifications = Switch(label="Enable notifications", default=True)
```

**Slider / DatePicker:**
```python
pain = Slider(label="Pain Level", min=0, max=10, step=1, default=5)
dob = DatePicker(label="Date of Birth", min_date="1900-01-01", max_date="2026-12-31")
```

**Hidden / FileField:**
```python
session_id = Hidden(label="Session", value="abc-123", required=False)
doc = FileField(label="Document", accept=["application/pdf", "image/*"], max_size=10_000_000)
```

#### Form

Single-step form container. Metadata is passed via `__init__` (all keyword-only):

```python
class Feedback(Form):
    comment = Text(label="Comment", multiline=True)

form = Feedback(
    title="Feedback",
    description="Tell us what you think",
    submit_label="Send Feedback",
    allow_cancel=True,
    cancel_label="Dismiss",
)
streamer.set_response_schema(form)
```

| Parameter | Type | Default | Description |
|---|---|---|---|
| `title` | `str` | required | Form heading |
| `description` | `str` | `None` | Subtitle/description text |
| `submit_label` | `str` | `"Submit"` | Submit button text |
| `allow_cancel` | `bool` | `False` | Show a cancel button |
| `cancel_label` | `str` | `"Cancel"` | Cancel button text |

#### MultiStepForm

Multi-step wizard. Define `Step` subclasses with fields, then use them as class attributes:

```python
from basion_agent.genui import MultiStepForm, Step, Text, Number, Slider, Checkbox

class Demographics(Step):
    name = Text(label="Full Name")
    age = Number(label="Age", min=0, max=150)

class PainAssessment(Step):
    pain_level = Slider(label="Pain Level", min=0, max=10)
    chronic = Checkbox(label="Is this chronic?")

class PatientIntake(MultiStepForm):
    demographics = Demographics(label="Basic Info", description="Your basic details")
    pain = PainAssessment(label="Pain Assessment")

form = PatientIntake(title="Patient Intake", description="Please complete all steps")
streamer.set_response_schema(form)
```

Field names must be **unique across all steps**. The frontend submits all fields as a **flat dict**.

#### Confirmation

A yes/no decision card — no input fields:

```python
from basion_agent.genui import Confirmation

confirm = Confirmation(
    title="Delete Account",
    message="Are you sure? This cannot be undone.",
    confirm_label="Delete",
    cancel_label="Keep Account",
    variant="destructive",  # "default", "destructive", "warning", "info"
)
streamer.set_response_schema(confirm)

# Parse the response
result = await Confirmation.parse(message)
result.confirmed  # bool — True if confirmed, False if cancelled
```

#### Parsing

`parse()` is always `async` and is a **classmethod**. It accepts `Message`, JSON string, or dict:

```python
data = await MyForm.parse(message)          # From Message (most common)
data = await MyForm.parse('{"name": "Alice"}')  # From JSON string
data = await MyForm.parse({"name": "Alice"})    # From dict

data.name       # str — typed access with IDE autocomplete
data.severity   # float
```

Missing fields get their type's default value. Extra keys are silently ignored.

For `FileField`, when parsing from a `Message`, the attachment client is automatically extracted:

```python
class UploadForm(Form):
    doc = FileField(label="Document")

data = await UploadForm.parse(message)
data.doc  # AttachmentInfo object
```

#### Form Cancellation

When a form has `allow_cancel=True`, the user can dismiss it and type a plain text message instead. The provider detects this automatically: if a `pending_response_schema` exists but the message content is not valid JSON, it marks the message with `formCancelled: True` in metadata.

```python
@agent.on_message
async def handle(message, sender):
    if message.schema:
        if message.form_cancelled:
            # User cancelled the form and typed plain text
            async with agent.streamer(message) as s:
                s.stream("No problem! How can I help you?")
            return

        # Normal form submission — parse and validate
        data = await SymptomReport.parse(message)
        result = SymptomReport.validate(data)
        if not result.is_valid:
            async with agent.streamer(message) as s:
                s.stream("Some fields are invalid:\n")
                for field_name, error in result.errors.items():
                    s.stream(f"- {field_name}: {error}\n")
            return
        # ... handle valid data
```

## Conversation History

```python
@agent.on_message
async def handle(message, sender):
    conv = message.conversation

    # Get message history
    history = await conv.get_history(limit=20)
    history = await conv.get_history(role="user", limit=10, offset=0)

    # Get conversation metadata
    metadata = await conv.get_metadata()

    # Get messages where this agent was sender or recipient
    agent_history = await conv.get_agent_history(limit=50)
    agent_history = await conv.get_agent_history(agent_name="triage-agent")
```

## Memory V2 (mem0)

Long-term memory powered by [mem0.ai](https://mem0.ai). Ingest messages and semantically search across a user's history.

```python
@agent.on_message
async def handle(message, sender):
    mem = message.memory_v2

    # Ingest the user's message
    await mem.ingest(role="user", content=message.content)

    # Search for relevant past context
    results = await mem.search(query="diagnosis history")
    for r in results:
        memory_text = r.get("memory")  # Extracted memory text

    async with agent.streamer(message) as s:
        if results:
            s.stream("Based on what I remember:\n")
            for r in results:
                s.stream(f"- {r.get('memory')}\n")
            s.stream("\n")
        s.stream(f"Regarding your question: {message.content}")

    # Ingest the assistant's response too
    await mem.ingest(role="assistant", content="...")
```

| Method | Description |
|---|---|
| `ingest(role, content)` | Ingest a message. Role: `'user'`, `'assistant'`, or `'system'`. |
| `search(query)` | Semantic search across the user's memories. Returns a list of dicts. |

## Attachments

Download and process file attachments (images, PDFs, etc.).

```python
@agent.on_message
async def handle(message, sender):
    if not message.has_attachments():
        return

    count = message.get_attachment_count()
    attachments = message.get_attachments()

    async with agent.streamer(message) as s:
        s.stream(f"Received {count} file(s).\n\n")

        for i, att in enumerate(attachments):
            s.stream(f"**{att.filename}** ({att.content_type}, {att.size} bytes)\n")

            if att.is_image():
                base64_str = await message.get_attachment_base64_at(i)
                # Send to vision model...

            elif att.is_pdf():
                pdf_bytes = await message.get_attachment_bytes_at(i)
                # Parse PDF...

        # Or download everything at once
        all_bytes = await message.get_all_attachment_bytes()
        all_base64 = await message.get_all_attachment_base64()
```

### Attachment Methods

| Method | Returns | Description |
|---|---|---|
| `has_attachments()` | `bool` | Whether the message has any attachments |
| `get_attachment_count()` | `int` | Number of attachments |
| `get_attachments()` | `List[AttachmentInfo]` | List of attachment metadata |
| `get_attachment_bytes()` | `bytes` | Download first attachment as bytes |
| `get_attachment_base64()` | `str` | Download first attachment as base64 |
| `get_attachment_buffer()` | `BytesIO` | Download first attachment as BytesIO |
| `get_attachment_bytes_at(i)` | `bytes` | Download attachment at index `i` |
| `get_attachment_base64_at(i)` | `str` | Download attachment at index `i` as base64 |
| `get_attachment_buffer_at(i)` | `BytesIO` | Download attachment at index `i` as BytesIO |
| `get_all_attachment_bytes()` | `List[bytes]` | Download all attachments |
| `get_all_attachment_base64()` | `List[str]` | Download all as base64 |
| `get_all_attachment_buffers()` | `List[BytesIO]` | Download all as BytesIO |

### AttachmentInfo Properties

| Property | Type | Example |
|---|---|---|
| `filename` | `str` | `"genetic_report.pdf"` |
| `content_type` | `str` | `"application/pdf"` |
| `size` | `int` | `524288` |
| `url` | `str` | Download URL |
| `file_extension` | `str` | `"pdf"` |
| `file_type` | `str` | `"pdf"` |
| `is_image()` | `bool` | `True` for image MIME types |
| `is_pdf()` | `bool` | `True` for PDF files |

## Structural Streaming

Rich UI components streamed alongside text. Bind a structural component to the streamer with `stream_by()`.

### Artifact

Files, images, or embeds with generation progress. Artifact data is persisted to the database.

```python
from basion_agent import Artifact

@agent.on_message
async def handle(message, sender):
    async with agent.streamer(message) as s:
        artifact = Artifact()

        # Show progress
        s.stream_by(artifact).generating("Generating genetic pathway diagram...", progress=0.3)
        # ... do work ...
        s.stream_by(artifact).generating("Rendering...", progress=0.8)

        # Complete with result
        s.stream_by(artifact).done(
            url="https://example.com/pathway-diagram.png",
            type="image",
            title="HTT Gene Pathway",
            description="Huntingtin protein interaction network",
            metadata={"width": 1200, "height": 800},
        )

        # Or signal an error
        # s.stream_by(artifact).error("Failed to generate diagram")

        s.stream("Here's the pathway diagram for the HTT gene.")
```

**Artifact types:** `image`, `iframe`, `document`, `video`, `audio`, `code`, `link`, `file`

### Surface

Interactive embedded components (iframes, widgets).

```python
from basion_agent import Surface

@agent.on_message
async def handle(message, sender):
    async with agent.streamer(message) as s:
        surface = Surface()
        s.stream_by(surface).generating("Loading appointment scheduler...")
        s.stream_by(surface).done(
            url="https://cal.com/embed/dr-smith",
            type="iframe",
            title="Schedule Genetic Counseling",
            description="Book a session with a genetic counselor",
        )
        s.stream("You can schedule your genetic counseling session above.")
```

### TextBlock

Collapsible text blocks with streaming title/body and visual variants. TextBlock events are **not** persisted.

```python
from basion_agent import TextBlock

@agent.on_message
async def handle(message, sender):
    async with agent.streamer(message) as s:
        block = TextBlock()

        # Set visual variant
        s.stream_by(block).set_variant("thinking")

        # Stream title (appends)
        s.stream_by(block).stream_title("Analyzing ")
        s.stream_by(block).stream_title("symptoms...")

        # Stream body (appends)
        s.stream_by(block).stream_body("Checking symptom database...\n")
        s.stream_by(block).stream_body("Cross-referencing with HPO ontology...\n")
        s.stream_by(block).stream_body("Matching against known phenotypes...\n")

        # Replace title/body entirely
        s.stream_by(block).update_title("Analysis Complete")
        s.stream_by(block).update_body("Found 3 matching conditions.")

        # Mark as done
        s.stream_by(block).done()

        s.stream("Based on the symptoms, here are possible conditions...")
```

**Variants:** `thinking`, `note`, `warning`, `error`, `success`

### Stepper

Multi-step progress indicators. Stepper events are **not** persisted.

```python
from basion_agent import Stepper

@agent.on_message
async def handle(message, sender):
    async with agent.streamer(message) as s:
        stepper = Stepper(steps=[
            "Search diseases",
            "Analyze phenotypes",
            "Find similar conditions",
            "Generate report",
        ])

        s.stream_by(stepper).start_step(0)
        diseases = await kg.search_diseases(name="Huntington")
        s.stream_by(stepper).complete_step(0)

        s.stream_by(stepper).start_step(1)
        phenotypes = await kg.search_phenotypes(name="chorea")
        s.stream_by(stepper).complete_step(1)

        s.stream_by(stepper).start_step(2)
        similar = await kg.find_similar_diseases("Huntington Disease")
        s.stream_by(stepper).complete_step(2)

        # Add a step dynamically
        s.stream_by(stepper).add_step("Cross-reference")
        s.stream_by(stepper).start_step(4)
        s.stream_by(stepper).update_step_label(4, "Cross-reference (final)")
        s.stream_by(stepper).complete_step(4)

        s.stream_by(stepper).start_step(3)
        # Or mark a step as failed
        # s.stream_by(stepper).fail_step(3, error="Report generation timed out")
        s.stream_by(stepper).complete_step(3)

        s.stream_by(stepper).done()
        s.stream("Here's your rare disease report...")
```

## Generative UI Message Components

Rich UI components that are **persisted as the message content**. Unlike structural streaming (which is ephemeral side-channel UI) and response schemas (which replace the chat input), genui message components are the actual message — stored in the database and rendered by the frontend as structured cards, accordions, etc.

### Overview

| Feature | Method | Persisted? | Renders As |
|---|---|---|---|
| Text streaming | `s.stream()` | Yes | Markdown text |
| Structural events | `s.stream_by()` | No (except Artifact) | Stepper, TextBlock, etc. |
| Response schemas | `s.set_response_schema()` | No | Form replacing chat input |
| **Genui components** | **`s.generate_ui()`** | **Yes** | **Card, Accordion, etc.** |

### Quick Start

```python
from basion_agent.genui import Card, CardSection, CardButton

@agent.on_message
async def handle(message, sender):
    async with agent.streamer(message) as s:
        s.generate_ui(Card(
            title="Lab Results",
            body="Blood work completed successfully.",
            variant="success",
            sections=[
                CardSection(label="Hemoglobin", value="14.2 g/dL"),
                CardSection(label="WBC", value="7,500 /uL"),
            ],
        ))
```

The message is stored with `is_genui=True` and `content` as a JSON array of component dicts. The frontend parses and renders them as structured UI instead of markdown.

### Card

A card component for displaying structured information with optional sections, image, and button.

```python
from basion_agent.genui import Card, CardSection, CardButton

Card(
    title="Patient Summary",                    # Required — card heading
    body="Overview of current status.",          # Optional — description text
    variant="info",                             # Optional — visual style
    sections=[                                  # Optional — key-value pairs
        CardSection(label="Name", value="John Doe"),
        CardSection(label="Age", value="42"),
        CardSection(label="Status", value="Active"),
    ],
    image="https://example.com/photo.jpg",      # Optional — image URL at top
    button=CardButton(                          # Optional — link button at bottom
        label="View Full Report",
        url="https://example.com/report/123",
    ),
)
```

| Parameter | Type | Default | Description |
|---|---|---|---|
| `title` | `str` | required | Card heading |
| `body` | `str` | `""` | Body text / description |
| `variant` | `str` | `"default"` | Visual style: `default`, `success`, `warning`, `error`, `info` |
| `sections` | `List[CardSection]` | `None` | Key-value pairs displayed in the card |
| `image` | `str` | `None` | Image URL displayed at the top of the card |
| `button` | `CardButton` | `None` | Link button at the bottom of the card |

**CardSection** — a key-value pair:

| Parameter | Type | Description |
|---|---|---|
| `label` | `str` | Label/key text |
| `value` | `str` | Value text |

**CardButton** — a link button:

| Parameter | Type | Description |
|---|---|---|
| `label` | `str` | Button text |
| `url` | `str` | URL the button opens |

### Accordion

A collapsible accordion component with expandable items.

```python
from basion_agent.genui import Accordion, AccordionItem

Accordion(
    title="Frequently Asked Questions",         # Required — heading
    items=[                                     # Required — at least one item
        AccordionItem(
            title="What is this service?",
            body="A multi-agent AI assistant for healthcare tasks.",
        ),
        AccordionItem(
            title="How do I submit a form?",
            body="Fill in the fields and press Submit.",
        ),
    ],
)
```

| Parameter | Type | Default | Description |
|---|---|---|---|
| `title` | `str` | required | Accordion heading |
| `items` | `List[AccordionItem]` | required | Expandable items (min 1) |

**AccordionItem**:

| Parameter | Type | Description |
|---|---|---|
| `title` | `str` | Item heading (clickable) |
| `body` | `str` | Content revealed when expanded |

### Multiple Components in One Message

Call `generate_ui()` multiple times to send several components as a single message:

```python
async with agent.streamer(message) as s:
    s.generate_ui(Card(
        title="Lab Results",
        body="Blood work completed on 2026-02-11.",
        variant="success",
        sections=[
            CardSection(label="Hemoglobin", value="14.2 g/dL (Normal)"),
            CardSection(label="WBC", value="7,500 /uL (Normal)"),
        ],
    ))
    s.generate_ui(Accordion(
        title="Detailed Breakdown",
        items=[
            AccordionItem(title="Complete Blood Count", body="All values within normal range."),
            AccordionItem(title="Metabolic Panel", body="Glucose: 95 mg/dL — normal."),
        ],
    ))
```

Both components are stored as a single message with `content = [card_dict, accordion_dict]`.

### Mutual Exclusion with `stream()`

`generate_ui()` and `stream()` cannot be mixed in the same message — they are mutually exclusive modes:

```python
# OK — genui only
async with agent.streamer(message) as s:
    s.generate_ui(Card(title="Result", body="Done."))

# OK — text only
async with agent.streamer(message) as s:
    s.stream("Here are the results in plain text...")

# ERROR — mixing modes raises RuntimeError
async with agent.streamer(message) as s:
    s.stream("Some text")
    s.generate_ui(Card(title="Oops"))  # RuntimeError!
```

### Combining with Structural Events

Structural events (`stream_by()`) work alongside `generate_ui()` because they are non-persisted side-channel UI:

```python
from basion_agent import Stepper
from basion_agent.genui import Card, CardSection

async with agent.streamer(message) as s:
    stepper = Stepper(steps=["Fetch data", "Analyze", "Build card"])

    s.stream_by(stepper).start_step(0)
    data = await fetch_data()
    s.stream_by(stepper).complete_step(0)

    s.stream_by(stepper).start_step(1)
    results = analyze(data)
    s.stream_by(stepper).complete_step(1)

    s.stream_by(stepper).start_step(2)
    s.generate_ui(Card(
        title="Analysis Complete",
        variant="success",
        sections=[
            CardSection(label="Records Analyzed", value=str(len(results))),
        ],
    ))
    s.stream_by(stepper).complete_step(2)
    s.stream_by(stepper).done()
```

### Raw Dict Fallback

If you need to send a component type not yet supported by the SDK classes, pass a raw dict:

```python
s.generate_ui({"type": "card", "title": "Custom", "body": "Raw dict works too."})
```

Any object with a `to_dict()` method or a plain dict is accepted.

### How It Works

1. Agent calls `s.generate_ui(component)` — component is appended to an internal list
2. On `finish()`, all components are serialized as a JSON array and sent as the message `content` with `isGenui=True`
3. Provider consumer stores the message with `is_genui=true` in the database
4. Frontend receives the Centrifugo event with `isGenui: true`, parses `content` as JSON, and renders the components

## Knowledge Graph

Query biomedical knowledge graphs for diseases, proteins, phenotypes, drugs, and pathways. Accessed via `agent.tools.knowledge_graph`.

```python
@agent.on_message
async def handle(message, sender):
    kg = agent.tools.knowledge_graph

    # Search diseases
    diseases = await kg.search_diseases(name="Huntington", limit=5)
    diseases = await kg.search_diseases(orphacode="399", limit=5)
    diseases = await kg.search_diseases(omim="143100")
    disease = await kg.get_disease(disease_id=123)

    # Search proteins/genes
    proteins = await kg.search_proteins(symbol="HTT", limit=10)
    proteins = await kg.search_proteins(ensembl_id="ENSG00000197386")

    # Search phenotypes (HPO terms)
    phenotypes = await kg.search_phenotypes(name="chorea", limit=10)
    phenotypes = await kg.search_phenotypes(hpo_id="HP:0002072")

    # Search drugs
    drugs = await kg.search_drugs(name="tetrabenazine", limit=5)

    # Search pathways
    pathways = await kg.search_pathways(name="apoptosis", limit=5)

    # Find similar diseases by shared phenotypes
    similar = await kg.find_similar_diseases("Huntington Disease", limit=10)
    for s in similar:
        s.disease_name        # "Spinocerebellar Ataxia Type 17"
        s.similarity_score    # 0.85
        s.shared_count        # 12

    # Find similar diseases by shared genes
    by_genes = await kg.find_similar_diseases_by_genes("Huntington Disease", limit=10)

    # Get all connections for an entity
    edges = await kg.get_entity_network("HTT", "protein")
    for e in edges:
        e.source_id, e.source_type
        e.target_id, e.target_type
        e.relation_type

    # k-hop graph traversal (BFS