Documentation Assistant Agent

A complete example demonstrating how to build a documentation assistant using the CAIP Agents SDK. This example showcases key SDK features including agent creation, thread management, tool registration, and streaming responses. Check Full Code Here

Overview

This example builds a conversational assistant that helps users navigate documentation and access platform resources. It demonstrates:

✅ Agent initialization and configuration
✅ Thread creation and management
✅ Custom tool registration
✅ Streaming responses
✅ Multiple interface options (CLI, Streamlit)
✅ Conversation persistence

Project Structure

caip-docs-agent/
├── .env.example          # Environment configuration template
├── agent.py              # Agent setup and tool definitions
├── main.py               # CLI interface for testing
├── app.py                # Basic Streamlit chat interface
├── app_with_context.py   # Streamlit with thread management
└── app_thread_management.py  # Advanced thread management

Prerequisites

Python 3.12 or Python 3.13 (recommended)
CAIP Agent ID (created via CAIP Portal)
CAIP API Key
Vector Store Key
CAIP Space ID

Setup

1. Environment Configuration

Create a .env file based on the example:

cd examples/caip-docs-agent
cp .env.example .env

Configure the required variables:

CAIP_API_KEY is a unified key that works for both Agents API and LLM API access. See CAIP API Key Authentication to obtain your key.

# SDK Configuration
CAIP_API_KEY=your_api_key_here
CAIP_SPACE_ID=your_space_id_here
CAIP_AGENT_ID=your_agent_id_here
CAIP_REGION=ROW                             # Optional: ROW (default) or CN

# Optional Configuration
MAX_HISTORY_MESSAGES=10                 # Messages to include in context
MAX_CONTEXT_TOKENS=1000                 # Token limit for context

2. Install Dependencies

pip install caip-agents-sdk streamlit

Core Components

Agent Configuration (`agent.py`)

The agent module sets up the SDK client, creates the agent, and registers tools.

"""
CAIP Documentation Agent - Configuration and Tools.

This module provides the core agent configuration and tool implementations
for the CAIP documentation assistant, including:
- Agent initialization and thread management
- Documentation search (hybrid vector + keyword)
- URL conversion for documentation links
- Resource link provisioning
"""

import logging
import os
from typing import Literal, Optional

from caip_agents_sdk import CAIPAgentsClient
from caip_agents_sdk.tools.knowledge_base import KnowledgeBaseSearchTool
from dotenv import load_dotenv

# ============================================================================
# Configuration
# ============================================================================

load_dotenv()

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Environment Configuration
class Config:
    """Centralized configuration management."""
    
    # API Configuration
    API_KEY = os.getenv("CAIP_API_KEY")
    BASE_URL = os.getenv("CAIP_LLM_EMBEDDINGS_URL", "https://llm.api.caip-test.bmw.cloud/v1")
    EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small")
    
    # Agent Configuration
    AGENT_ID = os.getenv("CAIP_AGENT_ID")
    SPACE_ID = os.getenv("CAIP_SPACE_ID")
    VECTOR_STORE_ID = os.getenv("VECTOR_STORE_ID")
    
    # Search Configuration
    TOP_K = int(os.getenv("TOP_K_RESULTS", "5"))
    VECTOR_WEIGHT = float(os.getenv("VECTOR_WEIGHT", "0.7"))  # 70% semantic, 30% keyword
    SCORE_THRESHOLD = float(os.getenv("SCORE_THRESHOLD", "0.0"))
    
    # CAIP Resource Links
    BASE_CONFLUENCE = "https://atc.bmwgroup.net/confluence/spaces/CONNECTAI/pages"
    LINKS = {
        "service_request": f"{BASE_CONFLUENCE}/7013030049/Raise+a+Service+Request",
        "incident": f"{BASE_CONFLUENCE}/7013030070/Raise+an+Incident",
        "feature_request": f"{BASE_CONFLUENCE}/7047749634/Raise+a+Feature+Request",
        "docs": "https://docs.caip.bmw.cloud/overview/",
        "dashboard": "https://caip.bmw.cloud/",
    }
    
    @classmethod
    def validate(cls) -> None:
        """Validate required configuration."""
        if not cls.API_KEY:
            logger.warning("CAIP_API_KEY is not set")
        if not cls.AGENT_ID:
            logger.error("CAIP_AGENT_ID is required but not set")
            raise ValueError("CAIP_AGENT_ID must be set in environment variables")
        if not cls.VECTOR_STORE_ID:
            logger.warning("VECTOR_STORE_ID is not set - search functionality may be limited")

# Validate configuration on module load
Config.validate()

logger.info(f"Configuration loaded:")
logger.info(f"  - Base URL: {Config.BASE_URL}")
logger.info(f"  - Embedding Model: {Config.EMBEDDING_MODEL}")
logger.info(f"  - Agent ID: {Config.AGENT_ID}")
logger.info(f"  - Vector Store ID: {Config.VECTOR_STORE_ID}")

# ============================================================================
# Client Initialization
# ============================================================================

# Initialize CAIP client
caip_client = CAIPAgentsClient()

# Create agent (thread_id set dynamically from app)
agent = caip_client.create_agent(
    framework="pydantic_ai",
    agent_id=Config.AGENT_ID,
)

# Initialize knowledge base search tool
kb_tool = KnowledgeBaseSearchTool(
    client=caip_client,
    vector_store_id=Config.VECTOR_STORE_ID
) if Config.VECTOR_STORE_ID else None

# ============================================================================
# Thread Management
# ============================================================================

def set_agent_thread(thread_id: str) -> None:
    """Set the agent's active thread ID.
    
    Args:
        thread_id: Thread identifier to set as active
    """
    agent.thread_id = thread_id
    logger.info(f"Agent thread ID updated to: {thread_id}")


def get_agent_thread() -> Optional[str]:
    """Get the current agent thread ID.
    
    Returns:
        Current thread ID or None if not set
    """
    return getattr(agent, 'thread_id', None)


async def create_new_thread(title: str = "New Chat") -> str:
    """Create a new conversation thread.
    
    Args:
        title: Thread title (generated from first message)
        
    Returns:
        Created thread ID
        
    Raises:
        Exception: If thread creation fails
    """
    try:
        thread_data = {"title": title, "status": "open"}
        thread = await caip_client.create_thread(
            agent_id=Config.AGENT_ID,
            thread_data=thread_data
        )
        logger.info(f"Created new thread: {thread.threadId} with title: {title}")
        return thread.threadId
    except Exception as e:
        logger.error(f"Failed to create thread: {e}")
        raise


# ============================================================================
# Agent Tools
# ============================================================================

@agent.tool_plain
def convert_to_docs_link(file_path: str) -> str:
    """Convert file path to CAIP documentation URL.

    Converts file system paths from the CAIP documentation repository
    to their corresponding URLs on https://docs.caip.bmw.cloud/

    IMPORTANT: Only works with actual FILE PATHS from docs/ directory.
    Rejects Docusaurus internal category paths like "category/documentation/contentType/...".

    Supported path patterns:
    - guides: 1-guides/caip-workflows/getting-started.md
    - features: 2-features/1-core-features/caip-apps/architecture.md
    - api-reference: 3-api-reference/agents-api-reference.md
    - platform: 4-platform-evolution/migrate-to-mkflow/migration.md
    - support: 5-support/incident.md
    - service-requests: 0-service-requests/index.md
    - root: index.md, faq.md, roadmap.md

    Args:
        file_path: File path from the docs/ directory

    Returns:
        Proper CAIP documentation URL or error message

    Examples:
        "1-guides/caip-workflows/getting-started.md" →
            "https://docs.caip.bmw.cloud/guides/caip-workflows/getting-started"
        "2-features/3-generativeai/llm-api/llm-api.md" →
            "https://docs.caip.bmw.cloud/features/generativeai/llm-api/"
    """
    logger.info(f"Converting file path to docs link: '{file_path}'")

    # Reject Docusaurus internal paths
    if file_path.startswith("category/documentation") or "contentType" in file_path:
        logger.warning(f"Rejected Docusaurus category path: {file_path}")
        return "❌ Invalid path - This looks like a Docusaurus internal category path, not a file path."

    # Normalize path
    clean_path = file_path.strip().rstrip("/").replace(".md", "").removeprefix("docs/")
    parts = [p for p in clean_path.split("/") if p]

    if not parts:
        return "https://docs.caip.bmw.cloud/"

    # Remove numeric prefixes (e.g., "1-guides" → "guides")
    processed_parts = []
    for part in parts:
        if part and part[0].isdigit() and "-" in part:
            cleaned = part[part.find("-") + 1:]
        else:
            cleaned = part
        
        if cleaned:
            processed_parts.append(cleaned)

    if not processed_parts:
        return "https://docs.caip.bmw.cloud/"
    
    # Remove "index" if it's the last part (index.md files should map to directory URLs)
    if processed_parts[-1] == "index":
        processed_parts = processed_parts[:-1]
        if not processed_parts:
            return "https://docs.caip.bmw.cloud/"

    # Remove duplicate filename (e.g., "llm-api/llm-api" → "llm-api/")
    add_trailing_slash = False
    if len(processed_parts) > 1 and processed_parts[-1] == processed_parts[-2]:
        processed_parts = processed_parts[:-1]
        add_trailing_slash = True

    first_part = processed_parts[0].lower()

    # Root-level documents
    if first_part in {"overview", "faq", "roadmap", "blog"} and len(processed_parts) == 1:
        return f"https://docs.caip.bmw.cloud/{first_part}"

    # Service requests (always at root)
    if first_part == "service-requests":
        path = "/".join(processed_parts)
        return f"https://docs.caip.bmw.cloud/{path}{'/' if len(processed_parts) > 1 else ''}"

    # Category sections (use /category/ prefix for root-level access)
    if first_part in {"api-reference", "platform-evolution", "support"}:
        if len(processed_parts) == 1:
            return f"https://docs.caip.bmw.cloud/category/{first_part}"
        path = "/".join(processed_parts)
        return f"https://docs.caip.bmw.cloud/category/{path}"

    # Direct sections (guides and features - NO /category/ prefix)
    if first_part in {"guides", "features"}:
        path = "/".join(processed_parts)
        if add_trailing_slash or len(processed_parts) >= 2:
            return f"https://docs.caip.bmw.cloud/{path}/"
        return f"https://docs.caip.bmw.cloud/{path}"

    # Default fallback
    path = "/".join(processed_parts)
    return f"https://docs.caip.bmw.cloud/{path}"



@agent.tool_plain
async def provide_link(
    link_type: Literal["service_request", "incident", "feature_request", "docs", "dashboard"]
) -> str:
    """Provide resource links for CAIP portal, documentation, or support.
    
    Args:
        link_type: Type of link to provide
        
    Returns:
        Formatted link with description
    """
    logger.info(f"Providing link: {link_type}")

    actions = {
        "service_request": "raise a service request",
        "incident": "raise an incident",
        "feature_request": "raise a feature request",
        "docs": "view the documentation",
        "dashboard": "access the CAIP portal",
    }

    action = actions.get(link_type, "access this resource")
    return f"You can {action} here: {Config.LINKS[link_type]}"


@agent.tool_plain
async def search_docs(query: str) -> str:
    """Search CAIP documentation using hybrid search (vector + keyword).
    
    Uses MongoDB hybrid search combining semantic similarity (vector embeddings)
    with keyword matching for optimal relevance.
    
    Args:
        query: Search query string
        
    Returns:
        Search results or error message
    """
    logger.info(f"Searching docs: '{query}'")

    # Validate configuration
    if not Config.VECTOR_STORE_ID:
        return "❌ Vector store not configured. Please set VECTOR_STORE_ID in .env"

    if not kb_tool:
        return "❌ Knowledge base search tool not initialized."

    try:
        # Perform hybrid search using SDK
        results = await kb_tool.search(
            query=query,
            search_type="hybrid",
            top_k=Config.TOP_K,
            vector_weight=Config.VECTOR_WEIGHT,
            score_threshold=Config.SCORE_THRESHOLD,
            metadata_filter=None,
            filter_logic="OR",
            document_id=None,
        )
        
        if not results:
            logger.info("No results from hybrid search")
            return "No relevant documentation found for your query."
        
        result_count = results.count("**Result")
        logger.info(f"Found {result_count} search results")
        return results
    
    except Exception as e:
        logger.error(f"Search failed: {e}", exc_info=True)
        return f"❌ Search failed: {str(e)}"

Key SDK Features Used:

CAIPAgentsClient() - Initialize the SDK client
create_agent() - Create a framework-specific agent
create_thread() - Create conversation threads
@agent.tool_plain - Register tools with decorators

CLI Interface (`main.py`)

A simple command-line interface for testing the agent.

"""CLI test script for CAIP Documentation Agent."""

import asyncio
from agent import agent, create_new_thread


async def main():
    """Test the CAIP Documentation Agent via CLI."""
    query = "How do I raise a service request?"
    
    print("🤖 CAIP Documentation Agent - CLI Test")
    print("=" * 50)
    print(f"Query: {query}\n")

    try:
        # Initialize the agent
        print("Initializing agent...")
        await agent.initialize()
        
        # Create a new thread
        thread_id = await create_new_thread("CLI Test")
        agent.thread_id = thread_id
        print("✅ Agent initialized successfully")

        # Stream the response
        print("\n🤖 Agent response:")
        print("-" * 30)
        
        stream = await agent.stream(query)
        async for chunk in stream.stream_text():
            print(chunk, end="", flush=True)

        print("\n" + "-" * 30)
        print("✅ Streaming complete!")

    except Exception as e:
        print(f"❌ Error: {e}")
        return 1

    return 0


if __name__ == "__main__":
    exit_code = asyncio.run(main())
    exit(exit_code)

Run the CLI:

python main.py

Key SDK Features Used:

agent.initialize() - Initialize agent connection
agent.stream() - Stream responses
stream.stream_text() - Iterate over response chunks

Basic Streamlit App (`app.py`)

A simple chat interface using Streamlit.

"""Streamlit chat interface for CAIP Documentation Agent."""

import asyncio
import os
from datetime import datetime

import streamlit as st
from agent import agent, create_new_thread, set_agent_thread
from dotenv import load_dotenv

# Load environment
load_dotenv()
MAX_HISTORY = int(os.getenv("MAX_HISTORY_MESSAGES", "10"))
MAX_TOKENS = int(os.getenv("MAX_CONTEXT_TOKENS", "1000"))
AGENT_ID = os.getenv("CAIP_AGENT_ID")

# Page config
st.set_page_config(page_title="CAIP Docs Assistant", page_icon="🤖")

st.title("🤖 CAIP Documentation Assistant")
st.caption("Ask about CAIP platform features, APIs, and capabilities")

# Initialize session state
if "threads" not in st.session_state:
    st.session_state.threads = {}
if "current_thread_id" not in st.session_state:
    st.session_state.current_thread_id = None

current_thread = None
if st.session_state.current_thread_id:
    current_thread = st.session_state.threads.get(st.session_state.current_thread_id)


def get_thread_title(message):
    """Generate a title for the thread based on first message."""
    if message:
        # Truncate to first 40 characters for a cleaner look
        title = message[:40] + "..." if len(message) > 40 else message
        return title
    return "New Chat"


# Get or create event loop
def get_event_loop():
    try:
        loop = asyncio.get_event_loop()
        if loop.is_closed():
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    return loop


# Initialize agent once
if "agent_initialized" not in st.session_state and agent._initialized is False:
    loop = get_event_loop()
    loop.run_until_complete(agent.initialize())
    # agent.add_tool_plain(provide_link)
    # agent.add_tool_plain(search_docs)
    st.session_state.agent_initialized = True

# Initialize message history
if "message_history" not in st.session_state:
    st.session_state.message_history = []

# Display chat history for current thread
if current_thread:
    for msg in current_thread["messages"]:
        with st.chat_message(msg["role"]):
            st.markdown(msg["content"])

# Handle user input
if prompt := st.chat_input("What would you like to know?"):
    if current_thread is None:
        loop = get_event_loop()
        try:
            thread_title = get_thread_title(prompt)
            new_thread_id = loop.run_until_complete(create_new_thread(thread_title))
            st.session_state.current_thread_id = new_thread_id
            st.session_state.threads[new_thread_id] = {
                "messages": [],
                "message_history": [],
                "created_at": datetime.now(),
                "title": thread_title,
            }
            # Update agent thread ID
            set_agent_thread(new_thread_id)
            current_thread = st.session_state.threads[new_thread_id]
        except Exception as e:
            st.error(f"Failed to create new thread: {e}")
            st.stop()

    # Add user message to current thread
    current_thread["messages"].append({"role": "user", "content": prompt})

    # Display user message
    with st.chat_message("user"):
        st.markdown(prompt)

    # Get agent response
    with st.chat_message("assistant"):
        placeholder = st.empty()

        async def stream_response():
            response = ""
            # Use thread-specific message history or fallback to global
            thread_history = current_thread.get("message_history", [])
            recent_history = (
                thread_history[-MAX_HISTORY:]
                if thread_history
                else st.session_state.message_history[-MAX_HISTORY:]
            )

            # Estimate tokens (rough: 1 token ≈ 4 chars) and truncate
            history_text = str(recent_history)
            if len(history_text) > MAX_TOKENS * 4:
                # Keep only recent messages that fit
                truncated = []
                char_count = 0
                for msg in reversed(recent_history):
                    msg_chars = len(str(msg))
                    if char_count + msg_chars > MAX_TOKENS * 4:
                        break
                    truncated.insert(0, msg)
                    char_count += msg_chars
                recent_history = truncated

            stream_result = await agent.stream(prompt, message_history=recent_history)

            async for chunk in stream_result.stream_text():
                response += chunk
                placeholder.markdown(response + "▌")
            placeholder.markdown(response)

            # Update message history - manually add user and assistant messages
            if "message_history" not in current_thread:
                current_thread["message_history"] = []
            
            # Add user message to history
            current_thread["message_history"].append({
                "role": "user",
                "content": prompt
            })
            st.session_state.message_history.append({
                "role": "user", 
                "content": prompt
            })
            
            # Add assistant response to history
            current_thread["message_history"].append({
                "role": "assistant",
                "content": response
            })
            st.session_state.message_history.append({
                "role": "assistant",
                "content": response
            })
            
            return response

        response = asyncio.run(stream_response())

    # Add assistant response to current thread
    current_thread["messages"].append({"role": "assistant", "content": response})

Run the Streamlit app:

streamlit run app.py

Key SDK Features Used:

Session-based thread management
Streaming with real-time UI updates
Message history for context

Streamlit with Thread Management (`app_with_context.py`)

An enhanced version with sidebar thread navigation and database persistence.

"""Streamlit chat interface for CAIP Documentation Agent with thread management."""

import asyncio
import os
from datetime import datetime

import streamlit as st
from agent import agent, caip_client, set_agent_thread, create_new_thread
from dotenv import load_dotenv

load_dotenv()
MAX_HISTORY = int(os.getenv("MAX_HISTORY_MESSAGES", "10"))
AGENT_ID = os.getenv("CAIP_AGENT_ID")

st.set_page_config(page_title="CAIP Docs Assistant", page_icon="🤖", layout="wide")


def get_event_loop():
    try:
        loop = asyncio.get_event_loop()
        if loop.is_closed():
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    return loop


def get_thread_title(message):
    """Generate a title from first message."""
    if message:
        return message[:40] + "..." if len(message) > 40 else message
    return "New Chat"


def extract_text_content(content_obj):
    """Extract text from various content formats."""
    if isinstance(content_obj, str):
        return content_obj
    if isinstance(content_obj, list) and content_obj:
        item = content_obj[0]
        if hasattr(item, "text"):
            return item.text
        return str(item)
    if hasattr(content_obj, "text"):
        return content_obj.text
    return str(content_obj)


async def load_threads_from_db():
    """Load thread metadata from database (lightweight, no messages)."""
    try:
        threads = await caip_client.list_threads(agent_id=AGENT_ID)
        threads_dict = {}

        for thread in threads:
            # Get title from thread object (API returns it as direct attribute)
            title = getattr(thread, "title", None)
            
            # Fallback checks for other possible locations
            if not title:
                if hasattr(thread, "data") and thread.data:
                    title = thread.data.get("title")
                elif hasattr(thread, "metadata") and thread.metadata:
                    title = thread.metadata.get("title")
            
            # Default to "New Chat" if no title found or empty
            if not title or not title.strip():
                title = "New Chat"

            try:
                created_at = (
                    datetime.fromisoformat(
                        thread.createdAt.replace("Z", "+00:00")
                    ).replace(tzinfo=None)
                    if hasattr(thread, "createdAt") and thread.createdAt
                    else datetime.now()
                )
            except (ValueError, AttributeError):
                created_at = datetime.now()

            threads_dict[thread.threadId] = {
                "title": title,
                "created_at": created_at,
                "messages": None,  # Lazy loaded - only when thread is clicked
                "messages_loaded": False,  # Track if messages have been loaded
            }

        return threads_dict
    except Exception as e:
        st.error(f"Failed to load threads: {e}")
        return {}


async def load_thread_messages(thread_id):
    """Load messages for a thread from database."""
    try:
        messages = await caip_client.get_messages(
            agent_id=AGENT_ID, thread_id=thread_id
        )
        return [
            {
                "role": getattr(msg, "role", "user"),
                "content": extract_text_content(getattr(msg, "content", str(msg))),
            }
            for msg in messages
        ]
    except Exception as e:
        st.error(f"Failed to load messages: {e}")
        return []


# Initialize session state
if "threads" not in st.session_state:
    st.session_state.threads = {}
if "current_thread_id" not in st.session_state:
    st.session_state.current_thread_id = None
if "agent_initialized" not in st.session_state:
    st.session_state.agent_initialized = False
if "threads_loaded" not in st.session_state:
    st.session_state.threads_loaded = False

# Initialize agent
if not st.session_state.agent_initialized:
    loop = get_event_loop()
    try:
        if not getattr(agent, "_initialized", False):
            loop.run_until_complete(agent.initialize())
        st.session_state.agent_initialized = True
    except Exception as e:
        st.error(f"Failed to initialize agent: {e}")

# Load threads from DB
if not st.session_state.threads_loaded:
    loop = get_event_loop()
    st.session_state.threads = loop.run_until_complete(load_threads_from_db())
    st.session_state.threads_loaded = True

# Sidebar
with st.sidebar:
    st.title("🤖 CAIP Docs Assistant")

    if st.button("📂 New chat", use_container_width=True):
        st.session_state.current_thread_id = None

    st.markdown("**Your Chats**")

    if st.session_state.threads:
        for thread_id, thread_data in sorted(
            st.session_state.threads.items(),
            key=lambda x: x[1]["created_at"],
            reverse=True,
        ):
            is_current = thread_id == st.session_state.current_thread_id
            if st.button(
                f"📝 {thread_data['title']}",
                key=f"thread_{thread_id}",
                use_container_width=True,
                type="primary" if is_current else "secondary",
                disabled=is_current,
            ):
                st.session_state.current_thread_id = thread_id
                set_agent_thread(thread_id)
                
                # Lazy load messages only when thread is clicked
                if not thread_data.get("messages_loaded", False):
                    with st.spinner("Loading messages..."):
                        thread_data["messages"] = get_event_loop().run_until_complete(
                            load_thread_messages(thread_id)
                        )
                        thread_data["messages_loaded"] = True
                
                st.rerun()
    else:
        st.caption("No conversations yet")

# Main chat
st.title("🤖 CAIP Documentation Assistant")
st.caption("Ask about CAIP platform features, APIs, and capabilities")

current_thread = (
    st.session_state.threads.get(st.session_state.current_thread_id)
    if st.session_state.current_thread_id
    else None
)

# Display chat history
if current_thread:
    # Ensure messages are loaded before displaying
    if not current_thread.get("messages_loaded", False):
        with st.spinner("Loading messages..."):
            current_thread["messages"] = get_event_loop().run_until_complete(
                load_thread_messages(st.session_state.current_thread_id)
            )
            current_thread["messages_loaded"] = True
    
    # Display messages if they exist
    if current_thread.get("messages"):
        for msg in current_thread["messages"]:
            with st.chat_message(msg["role"]):
                st.markdown(msg["content"])

# Handle input
if prompt := st.chat_input("What would you like to know?"):
    loop = get_event_loop()

    # Create new thread if needed
    if current_thread is None:
        try:
            thread_title = get_thread_title(prompt)
            new_thread_id = loop.run_until_complete(create_new_thread(thread_title))
            st.session_state.current_thread_id = new_thread_id
            st.session_state.threads[new_thread_id] = {
                "messages": [],
                "created_at": datetime.now(),
                "title": thread_title,
                "messages_loaded": True,  # New threads start with loaded state
            }
            set_agent_thread(new_thread_id)
            current_thread = st.session_state.threads[new_thread_id]
        except Exception as e:
            st.error(f"Failed to create thread: {e}")
            st.stop()

    # Display user message
    current_thread["messages"].append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    # Get and display response with streaming
    with st.chat_message("assistant"):
        placeholder = st.empty()

        async def stream_response():
            response = ""
            try:
                # Get message history
                message_history = await agent.list_messages(
                    thread_id=st.session_state.current_thread_id,
                    limit=MAX_HISTORY,
                )
                
                # Stream the response
                stream_result = await agent.stream(prompt, message_history=message_history)
                
                # Display streaming text with cursor
                async for chunk in stream_result.stream_text():
                    response += chunk
                    placeholder.markdown(response + "▌")
                
                # Display final response without cursor
                placeholder.markdown(response)
                return response
                
            except Exception as e:
                error_msg = f"Error: {e}"
                placeholder.markdown(error_msg)
                return error_msg

        response = loop.run_until_complete(stream_response())

    current_thread["messages"].append({"role": "assistant", "content": response})
    st.rerun()

Key SDK Features Used:

caip_client.list_threads() - Load threads from backend
agent.list_messages() - Load message history from backend
Thread switching with context restoration

Running the Example

Option 1: CLI Test

cd examples/caip-docs-agent
python main.py

Option 2: Basic Chat UI

cd examples/caip-docs-agent
streamlit run app.py

Option 3: Full Thread Management UI

cd examples/caip-docs-agent
streamlit run app_with_context.py

Key SDK Patterns Demonstrated

1. Agent Initialization

from caip_agents_sdk import CAIPAgentsClient

client = CAIPAgentsClient()
agent = client.create_agent(framework="pydantic_ai", agent_id=AGENT_ID)
await agent.initialize()

2. Dynamic Thread Management

# Create thread
thread = await client.create_thread(
    agent_id=AGENT_ID,
    thread_data={"title": "My Conversation"}
)

# Assign to agent
agent.thread_id = thread.threadId

3. Tool Registration

@agent.tool_plain
def my_tool(param: str) -> str:
    """Tool description for the agent."""
    return f"Result: {param}"

4. Streaming Responses

stream = await agent.stream(query, message_history=history)
async for chunk in stream.stream_text():
    print(chunk, end="", flush=True)

5. Loading Thread History

# List all threads
threads = await client.list_threads(agent_id=AGENT_ID)

# Load messages for a thread
messages = await agent.list_messages(
    thread_id=thread_id
)

Customization

Adding Custom Tools

Extend the agent with your own tools in agent.py:

@agent.tool_plain
def search_knowledge_base(query: str) -> str:
    """Search your knowledge base for relevant information."""
    # Your search implementation
    results = your_search_function(query)
    return format_results(results)

@agent.tool_plain
def get_user_info(user_id: str) -> str:
    """Get information about a user."""
    user = database.get_user(user_id)
    return f"User: {user.name}, Email: {user.email}"

Changing the Framework

Switch from PydanticAI to LangChain:

agent = caip_client.create_agent(
    framework="langchain",  # Changed from "pydantic_ai"
    agent_id=AGENT_ID,
)

Customizing Resource Links

Update the LINKS dictionary for your organization:

LINKS = {
    "docs": "https://docs.your-company.com/",
    "support": "https://support.your-company.com/",
    "dashboard": "https://app.your-company.com/",
}

Next Steps

Email Agent Example - Build an email automation agent
Tools Documentation - Learn more about tool registration
Threads Documentation - Deep dive into thread management

Overview​

Project Structure​

Prerequisites​

Setup​

1. Environment Configuration​

2. Install Dependencies​

Core Components​

Agent Configuration (agent.py)​

CLI Interface (main.py)​

Basic Streamlit App (app.py)​

Streamlit with Thread Management (app_with_context.py)​

Running the Example​

Option 1: CLI Test​

Option 2: Basic Chat UI​

Option 3: Full Thread Management UI​

Key SDK Patterns Demonstrated​

1. Agent Initialization​

2. Dynamic Thread Management​

3. Tool Registration​

4. Streaming Responses​

5. Loading Thread History​

Customization​

Adding Custom Tools​

Changing the Framework​

Customizing Resource Links​

Next Steps​

Overview

Project Structure

Prerequisites

Setup

1. Environment Configuration

2. Install Dependencies

Core Components

Agent Configuration (`agent.py`)

CLI Interface (`main.py`)

Basic Streamlit App (`app.py`)

Streamlit with Thread Management (`app_with_context.py`)

Running the Example

Option 1: CLI Test

Option 2: Basic Chat UI

Option 3: Full Thread Management UI

Key SDK Patterns Demonstrated

1. Agent Initialization

2. Dynamic Thread Management

3. Tool Registration

4. Streaming Responses

5. Loading Thread History

Customization

Adding Custom Tools

Changing the Framework

Customizing Resource Links

Next Steps