Agents

Agents are the core building blocks of the CAIP Agents SDK. Think of an agent as an AI teammate for your app: it understands your instructions, remembers conversation context, and uses tools to complete real tasks.

---

What is an Agent?

In the CAIP ecosystem, an Agent consists of two parts:

1. Backend Configuration (stored in CAIP Agents API)

   • Agent ID, name, and description
   • Model selection (GPT-4, Claude, etc.)
   • System instructions and prompts
   • Default parameters (temperature, max_tokens, etc.)
   • Provider configuration
   • Metadata

2. SDK Client (your application code)

   • Framework adapter (PydanticAI, LangChain, or LangGraph)
   • Tool registrations
   • Conversation management
   • Message handling

┌─────────────────────────────────────────────────────────┐
│                    CAIP Portal                          │
│  (Create & configure agents with model, instructions)   │
└──────────────────────────┬──────────────────────────────┘
                           │
                           ▼ Agent ID
┌─────────────────────────────────────────────────────────┐
│                 CAIP Agents SDK                         │
│  (Connect to agent, add tools, manage conversations)    │
└─────────────────────────────────────────────────────────┘

---

Agent Lifecycle

1. Create Agent in Portal

First, create your agent through the CAIP Portal. This is where you configure:

Property	Description	Example
name	Human-readable agent name	"Customer Support Agent"
instructions	System instructions for behavior	"You are a helpful assistant..."
description	Agent description/system prompt	"AI assistant for customer support"
model	LLM model to use	"gpt-4", "claude-3-sonnet"
provider	Model provider	"openai", "anthropic"
outputFormat	Response format	"text", "json"
defaultParameters	Model parameters	{"temperature": 0.7, "max_tokens": 1024}
metadata	Custom metadata	{"type": "conversational"}

After creation, you'll receive an Agent ID (e.g., 507f1f77bcf86cd799439011).

2. Connect via SDK

Use the Agent ID in your application:

from caip_agents_sdk import CAIPAgentsClient

# Initialize the main client
client = CAIPAgentsClient()

# Create a framework-specific agent client
agent = client.create_agent(
    framework="langchain",  # or "pydantic_ai" / "langgraph"
    agent_id="your-agent-id"
)

3. Initialize the Agent

The initialize() method connects to the backend and prepares the agent:

await agent.initialize()

What happens during initialization:

• Fetches agent configuration from CAIP Agents API
• Loads model settings (temperature, max_tokens, etc.)
• Creates the underlying framework client (LangChain, PydanticAI, or LangGraph)
• Configures the LLM with proper credentials and endpoints
• Registers any pending tools

4. Create a Conversation Thread

Before chatting, create a thread to store the conversation:

thread = await client.create_thread(
    agent_id="your-agent-id",
    thread_data={
        "title": "Customer Support Session",
        "status": "open"
    }
)

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

5. Run the Agent

Now you can interact with the agent:

# Simple query
response = await agent.run("Hello! How can you help me?")
print(response)

# With streaming
stream_result = await agent.stream("Tell me about your capabilities")
async for chunk in stream_result.stream_text():
    print(chunk, end="", flush=True)

# With iter (PydanticAI only) — fine-grained node-level streaming
async with await agent.iter("Tell me about your capabilities") as iter_result:
    async for node in iter_result:
        if iter_result.is_model_request_node(node):
            async with iter_result.stream_node(node) as stream:
                async for chunk in stream.stream_text(delta=False):
                    print(chunk, end="", flush=True)
            print()

print(f"Output: {iter_result.output}")

---

Supported Frameworks

The SDK supports three AI frameworks through a unified interface:

PydanticAI

agent = client.create_agent(
    framework="pydantic_ai",  
    agent_id="your-agent-id"
)

Features:

• Type-safe responses with Pydantic models
• Built on async/await patterns
• Structured output validation
• Native tool support with context

LangChain

agent = client.create_agent(
    framework="langchain",  
    agent_id="your-agent-id"
)

Features:

• Extensive ecosystem of tools and integrations
• ReAct agent pattern
• Memory and chain composition
• Wide model support

LangGraph

agent = client.create_agent(
    framework="langgraph",
    agent_id="your-agent-id"
)

Features:

• Build graph-based and multi-agent workflows
• Route dynamically between specialist nodes
• Use checkpointing for pause/resume and state inspection
• Keep SDK-managed persistence, logging, and observability

For a full step-by-step workflow, see Multi-Agentic Systems.

---

Agent Configuration

Default Parameters

Default model parameters are optional. You can add them in the CAIP self-service portal only when your use case needs custom behavior:

# Optional parameters configure in the self-service portal (only if needed)
{
    "temperature": 0.7,      # Creativity (0.0 - 2.0)
    "max_tokens": 1024,      # Maximum response length
    "top_p": 0.9,            # Nucleus sampling
    "frequency_penalty": 0.0, # Reduce repetition
    "presence_penalty": 0.0   # Encourage new topics
}

Model Override

You can override the model at runtime:

agent = client.create_agent(
    framework="pydantic_ai",
    agent_id="your-agent-id",
    model_name_override="gpt-4-turbo"  # Override portal setting
)

---

Iterating Agent Execution (PydanticAI only)

The iter() method provides node-level access to an agent run, letting you inspect and stream each step of the execution graph as it happens. This is useful when you need fine-grained control — for example, streaming text exactly as the model generates it while still being able to react to intermediate tool calls.

note

iter() is currently supported for the PydanticAI framework only.

How it works

agent.iter() returns an async context manager. Inside it, you iterate over nodes — each node represents one step in the agent's execution (e.g., a model request, a tool call, or a final result).

Helper	Description
iter_result.is_model_request_node(node)	Returns True if the node is a model generation step
iter_result.stream_node(node)	Opens a streaming context for the given node
stream.stream_text(delta=False)	Yields text chunks; delta=False yields the full accumulated text so far
iter_result.output	The final output string, available after iteration completes

Example

async with await agent.iter("what is the weather in delhi") as iter_result:
    async for node in iter_result:
        if iter_result.is_model_request_node(node):
            async with iter_result.stream_node(node) as stream:
                async for chunk in stream.stream_text(delta=False):
                    print(chunk, end="", flush=True)
            print()

print(f"Output: {iter_result.output}")

stream_text delta modes

delta value	Behaviour
False (default)	Each chunk contains the full accumulated text up to that point
True	Each chunk contains only the new text added since the last chunk

---

Complete Example

import asyncio
from caip_agents_sdk import CAIPAgentsClient

async def main():
    # 1. Initialize the SDK client
    client = CAIPAgentsClient()
    
    # 2. Create agent with your Agent ID from the portal
    agent = client.create_agent(
        framework="langchain",
        agent_id="your-agent-id"
    )
    
    # 3. Initialize (fetches config from backend)
    await agent.initialize()
    
    # 4. Create a conversation thread
    thread = await client.create_thread(
        agent_id="your-agent-id",
        thread_data={"title": "Demo Session", "status": "open"}
    )
    agent.thread_id = thread.threadId
    
    # 5. Register custom tools (optional)
    @agent.tool_plain
    def get_weather(city: str) -> str:
        """Get weather for a city."""
        return f"The weather in {city} is sunny, 22°C"
    
    # 6. Chat with the agent
    response = await agent.run("What's the weather in Munich?")
    print(f"Agent: {response}")
    
if __name__ == "__main__":
    asyncio.run(main())

---

Agent Methods Reference

Method	Description	Frameworks
initialize()	Connect to backend and prepare agent	All
run(query)	Send a message and get a response	All
stream(query)	Stream a response in chunks	All
iter(query)	Iterate over execution nodes with fine-grained streaming	PydanticAI only
resume(value)	Resume interrupted graph execution	LangGraph
get_state()	Read checkpointed graph state	LangGraph
tool(func)	Register a tool with context	All
tool_plain(func)	Register a tool without context	All
add_tool(func)	Programmatically add a tool	All
get_native_agent()	Access underlying framework agent	All

---

Best Practices

1. Always Initialize Before Use

agent = client.create_agent(framework="langchain", agent_id="...")
await agent.initialize()  # Required before run() or stream()

2. Use Descriptive Thread Titles

thread = await client.create_thread(
    agent_id="...",
    thread_data={
        "title": "Order #12345 Support",  # Helps identify conversations
        "status": "open"
    }
)

3. Handle Errors Gracefully

from caip_agents_sdk import CAIPAgentsClient, ValidationError, CAIPException

try:
    response = await agent.run("Hello")
except ValidationError as e:
    print(f"Invalid input: {e}")
except CAIPException as e:
    print(f"SDK error: {e}")

4. Choose the Right Framework

• PydanticAI: Best for type-safe applications, structured outputs
• LangChain: Best for complex workflows, extensive tool ecosystems
• LangGraph: Best for multi-agent workflows, conditional routing, checkpointing, and human-in-the-loop orchestration

---

Next Steps

• Threads - Learn about conversation management
• Tools - Add custom capabilities to your agents