Threads

Threads are conversation containers that store and manage the history of interactions between users and agents. They enable multi-turn conversations with persistent context across sessions.

---

What is a Thread?

A Thread represents a single conversation session with an agent. Think of it like a chat window or support ticket that:

• Stores all messages exchanged between the user and agent
• Persists across sessions so users can resume conversations
• Tracks metadata like timestamps and custom properties

┌─────────────────────────────────────────────────────────┐
│                     Thread                              │
│  ┌───────────────────────────────────────────────────┐  │
│  │ Message 1: User: "Hello, I need help"             │  │
│  │ Message 2: Agent: "Hi! How can I assist you?"     │  │
│  │ Message 3: User: "What's my order status?"        │  │
│  │ Message 4: Agent: "Let me check that for you..."  │  │
│  │ ...                                               │  │
│  └───────────────────────────────────────────────────┘  │
│                                                         │
│  Metadata: {priority: "high", category: "support"}      │
└─────────────────────────────────────────────────────────┘

---

Thread Properties

Property	Type	Description
threadId	string	Unique identifier for the thread
agentId	string	Associated agent ID
title	string (optional)	Human-readable title
metadata	object (optional)	Custom metadata
createdAt	datetime	Creation timestamp
updatedAt	datetime	Last update timestamp
lastMessageAt	datetime (optional)	Timestamp of last message

---

Creating Threads

Basic Thread Creation

from caip_agents_sdk import CAIPAgentsClient

client = CAIPAgentsClient()

# Create a new thread
thread = await client.create_thread(
    agent_id="your-agent-id",
    thread_data={
        "title": "Your Thread Title"
    }
)

print(f"Thread created: {thread.threadId}")

Thread with Metadata

thread = await client.create_thread(
    agent_id="your-agent-id",
    thread_data={
        "title": "Your Thread Title",
        "metadata": {
            "your_key": "your_value",
            "another_key": "another_value"
        }
    }
)

---

Assigning Threads to Agents

After creating a thread, assign it to your agent before running queries:

# Create agent
agent = client.create_agent(
    framework="pydantic_ai",
    agent_id="your-agent-id"
)
await agent.initialize()

# Create and assign thread
thread = await client.create_thread(
    agent_id="your-agent-id",
    thread_data={"title": "Your Thread Title"}
)

# Assign thread to agent
agent.thread_id = thread.threadId

# Now messages will be stored in this thread
response = await agent.run("Hello!")

---

Retrieving Threads

Get a Specific Thread

thread = await client.get_thread(
    agent_id="your-agent-id",
    thread_id="thread-id-here"
)

print(f"Title: {thread.title}")
print(f"Created: {thread.createdAt}")

List All Threads for an Agent

# Get all threads
threads = await client.list_threads(agent_id="your-agent-id")

for thread in threads:
    print(f"- {thread.title} (created: {thread.createdAt})")

# With pagination
threads = await client.list_threads(
    agent_id="your-agent-id",
    limit=10,
    offset=0
)

Working with Message History

How Message Storage Works

When you call agent.run() or agent.stream():

1. ✅ User message is stored in the thread (backend database)
2. ✅ Agent response is stored in the thread (backend database)
3. ❌ Previous messages are NOT automatically loaded for context

Important: The SDK stores messages for persistence but does not automatically fetch previous messages to provide conversation context to the agent. You must explicitly pass message_history to maintain context.

Providing Context with Message History

To give the agent context from previous messages, you need to:

1. Retrieve messages from the thread
2. Pass them as message_history to run() or stream()

# First message - no history needed
response1 = await agent.run("Your first message")

# Get previous messages for context
message_history = await agent.list_messages(
    thread_id=thread.threadId,
    limit=50  # Adjust based on your needs
)

# Second message - pass history for context
response2 = await agent.run(
    "Your follow-up question",
    message_history=message_history  # Agent now knows the context
)

Retrieve Messages from a Thread

messages = await client.get_messages(
    agent_id="agent-id-here",
    thread_id="thread-id-here",
    limit=50,
    offset=0
)

for msg in messages:
    print(f"{msg.role}: {msg.content}")

---

Thread Management Patterns

Pattern 1: Conversation with Context

async def chat_with_context(agent, thread_id: str, user_input: str):
    """Send a message with full conversation history."""
    
    # Get previous messages for context
    message_history = await agent.list_messages(
        thread_id=thread_id,
        limit=50  # Adjust based on your needs
    )
    
    # Run with context
    response = await agent.run(user_input, message_history=message_history)
    return response

Pattern 2: One Thread Per User Session

async def handle_user_session(user_id: str, agent_id: str):
    client = CAIPAgentsClient()
    agent = client.create_agent(framework="pydantic_ai", agent_id=agent_id)
    await agent.initialize()
    
    # Create a new thread for this session
    thread = await client.create_thread(
        agent_id=agent_id,
        thread_data={
            "title": f"Session for {user_id}",
            "metadata": {"user_id": user_id}
        }
    )
    agent.thread_id = thread.threadId
    
    return agent

Pattern 3: Resume Existing Thread with Context

async def resume_conversation(agent_id: str, thread_id: str, new_message: str):
    client = CAIPAgentsClient()
    agent = client.create_agent(framework="pydantic_ai", agent_id=agent_id)
    await agent.initialize()
    
    # Use existing thread
    agent.thread_id = thread_id
    
    # Get previous messages for context
    message_history = await agent.list_messages(
        thread_id=thread_id,
        limit=50  # Adjust based on your needs
    )
    
    # Continue conversation with full context
    response = await agent.run(new_message, message_history=message_history)
    return response

Pattern 4: Thread Selection UI (Streamlit Example)

import streamlit as st
from caip_agents_sdk import CAIPAgentsClient

async def thread_selector(agent_id: str):
    client = CAIPAgentsClient()
    
    # Load all threads
    threads = await client.list_threads(agent_id=agent_id)
    
    # Create selection options
    thread_options = {
        thread.title: thread.threadId 
        for thread in threads
    }
    
    # Sidebar selection
    selected = st.sidebar.selectbox(
        "Select Conversation",
        options=["New Chat"] + list(thread_options.keys())
    )
    
    if selected == "New Chat":
        thread = await client.create_thread(
            agent_id=agent_id,
            thread_data={"title": "Your Thread Title"}
        )
        return thread.threadId
    else:
        return thread_options[selected]

---

Complete Example

import asyncio
from caip_agents_sdk import CAIPAgentsClient

async def main():
    client = CAIPAgentsClient()
    agent_id = "your-agent-id"
    
    # Create agent
    agent = client.create_agent(framework="pydantic_ai", agent_id=agent_id)
    await agent.initialize()
    
    # Create a new conversation thread
    thread = await client.create_thread(
        agent_id=agent_id,
        thread_data={
            "title": "Your Thread Title",
            "metadata": {"your_key": "your_value"}
        }
    )
    print(f"Created thread: {thread.threadId}")
    
    # Assign to agent
    agent.thread_id = thread.threadId
    
    # First message
    response1 = await agent.run("Your first message here")
    print(f"Agent: {response1}")
    
    # Get previous messages for context
    message_history = await agent.list_messages(
        thread_id=thread.threadId,
        limit=50  # Adjust based on your needs
    )
    
    # Second message with context
    response2 = await agent.run(
        "Your second message here",
        message_history=message_history
    )
    print(f"Agent: {response2}")  # Agent remembers previous context

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

---

Best Practices

1. Use Descriptive Titles

# ❌ Bad
thread_data={"title": "Chat"}

# ✅ Good
thread_data={"title": f"Order #{order_id} Support - {customer_name}"}

2. Track Important Metadata

thread_data={
    "title": "Your Thread Title",
    "metadata": {
        "your_custom_key": "your_custom_value",
        "another_key": "another_value"
    }
}

3. Handle Thread Not Found

from caip_agents_sdk import NotFoundError

try:
    thread = await client.get_thread(agent_id, thread_id)
except NotFoundError:
    # Create a new thread if not found
    thread = await client.create_thread(
        agent_id=agent_id,
        thread_data={"title": "Your Thread Title"}
    )

---

Next Steps

• Messages - Learn about message structure and handling
• Tools - Add custom capabilities to your agents
• Examples - See threads in action