Skip to main content
VoltAgent DocsVoltOps DocsExamplesChangelog
SDKs

Python SDK

Track your AI agents with full observability - traces, sub-agents, tools, memory operations, and more.

Installation​

pip install voltagent

Setup​

Initialize the SDK with your credentials:

import asyncio
from voltagent import VoltAgentSDK

async def main():
    sdk = VoltAgentSDK(
        base_url="https://api.voltagent.dev",
        public_key="your-public-key",
        secret_key="your-secret-key",
        auto_flush=True,  # Auto-send events
        flush_interval=3,  # Send every 3 seconds
        timeout=30,
    )

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

Before using the SDK, you need to create an account at https://console.voltagent.dev/ and set up an organization and project to get your API keys.

Two Usage Approaches​

The Python SDK supports two approaches for resource management:

Automatic resource management with async context managers:

# Automatic resource management with async context managers
async with sdk.trace(
    agentId="my-agent",
    input={"query": "example"}
) as trace:
    # trace is automatically ended when exiting the context
    pass

Benefits:

  • Automatic cleanup and error handling
  • Pythonic and clean code
  • Exception safety
  • Recommended for most use cases

Step-by-Step Guide​

Create a Trace​

A trace represents one complete agent execution session. Every agent operation must happen within a trace.

async with sdk.trace(
    agentId="support-agent-v1",
    input={"query": "How to reset password?"},
    userId="user-123",
    conversationId="conv-456",
    tags=["support", "password-reset"],
    metadata={
        "priority": "high",
        "source": "web-chat",
    },
) as trace:
    print(f"Trace created: {trace.id}")
    # Trace automatically ends when exiting context

Trace creation in dashboard

Add an Agent to the Trace​

Now let's add the main agent that will handle the user's request:

agent = await trace.add_agent({
    "name": "Support Agent",
    "input": {"query": "User needs password reset help"},
    "instructions": "You are a customer support agent specialized in helping users with account issues and password resets.",
    "metadata": {
        "modelParameters": {
            "model": "gpt-4",
        },
    },
})

Trace with first agent

Understanding Agent Metadata​

Agent metadata helps you organize and filter your observability data. Here's what each field means:

  • modelParameters: Built-in SDK field for model configuration (model name, temperature, etc.)
  • Custom fields: Add any domain-specific metadata for your use case (user-defined)
metadata = {
    # Built-in SDK field
    "modelParameters": {
        "model": "gpt-4",
        "temperature": 0.7,
        "max_tokens": 1000,
    },
    # Your own custom metadata (examples)
    "user_session": "session_abc123",
    "priority": "high",
    "department": "customer-success",
    "region": "us-east-1",
}

Agent metadata in dashboard

💡 Trace Completion

async with sdk.trace(agentId="agent") as trace:
    # Trace automatically ends with SUCCESS when exiting normally
    # Automatically ends with ERROR if exception occurs
    pass

Add a Tool to the Agent​

Tools represent external services or APIs that your agent uses. Let's add a knowledge base search tool:

search_tool = await agent.add_tool({
    "name": "knowledge-base-search",
    "input": {
        "query": "password reset procedure",
        "max_results": 5,
    },
    "metadata": {
        # Add your own custom metadata
        "search_type": "semantic",
        "database": "support-kb",
        "version": "v2",
    },
})

Agent with tools

Tool Success​

When the tool executes successfully:

await search_tool.success(
    output={
        "results": ["Reset via email", "Reset via SMS", "Contact support"],
        "count": 3,
        "relevance_score": 0.89,
    },
    metadata={
        "search_time": "0.2s",
        "index_used": "support-kb-v2",
    },
)

Agent with tools success

Tool Error​

When the tool fails, you can report errors in two ways:

await search_tool.error(
    status_message=Exception("Database connection timeout"),
    metadata={
        "database": "support-kb",
        "timeout_ms": 5000,
    },
)

Agent with tools error

Add Memory Operations​

Memory operations track data storage and retrieval. They work exactly like tools with success and error states:

memory_op = await agent.add_memory({
    "name": "user-context-storage",
    "input": {
        "key": "user_123_context",
        "value": {
            "last_login": "2024-01-15",
            "account_type": "premium",
            "preferences": {"language": "en"},
        },
        "ttl": 3600,  # 1 hour
    },
    "metadata": {
        "type": "redis",
        "region": "us-east-1",
    },
})

Agent with memory

Memory Success​

await memory_op.success(
    output={
        "stored": True,
        "key": "user_123_context",
        "expires_at": "2024-01-15T15:00:00Z",
    },
    metadata={
        "cache_hit": False,
        "storage_latency": "2ms",
    },
)

Agent with memory success

Memory Error​

await memory_op.error(
    status_message=Exception("Redis connection failed"),
    metadata={
        "storage_type": "redis",
        "error_code": "CONNECTION_TIMEOUT",
    },
)

Add Retrieval Operations​

Retrievers handle data retrieval from vector stores, databases, or knowledge bases. They also follow the same success/error pattern:

retriever = await agent.add_retriever({
    "name": "policy-document-retriever",
    "input": {
        "query": "password reset policy for premium users",
        "max_documents": 3,
        "threshold": 0.8,
    },
    "metadata": {
        "vector_store": "pinecone",
        "embedding_model": "text-embedding-ada-002",
    },
})

Agent with retriever

Retriever Success​

await retriever.success(
    output={
        "documents": [
            "Premium users can reset passwords instantly via email",
            "Password reset requires 2FA verification for premium accounts",
            "Premium users have 24/7 phone support for password issues",
        ],
        "relevance_scores": [0.95, 0.88, 0.82],
    },
    metadata={
        "search_time": "0.3s",
        "documents_scanned": 1500,
    },
)

Retriever Error​

await retriever.error(
    status_message=Exception("Vector store unavailable"),
    metadata={
        "vector_store": "pinecone",
        "error_type": "SERVICE_UNAVAILABLE",
    },
)

Working with Sub-Agents​

Sub-agents create hierarchical agent structures. Each sub-agent can have its own tools, memory operations, and even more sub-agents:

# Create a sub-agent under the main agent
policy_checker = await agent.add_agent({
    "name": "Policy Checker",
    "input": {
        "user_id": "user-123",
        "request_type": "password-reset",
    },
    "instructions": "You are responsible for verifying customer requests against company policies.",
    "metadata": {
        "role": "policy-verification",
        "modelParameters": {
            "model": "gpt-4",
        },
    },
})

Agent with subagents

Sub-Agent Success​

from voltagent import TokenUsage

await policy_checker.success(
    output={
        "policy_compliant": True,
        "required_verification": "2fa-sms",
        "approval_granted": True,
    },
    usage=TokenUsage(
        prompt_tokens=85,
        completion_tokens=45,
        total_tokens=130,
    ),
    metadata={
        "policies_checked": ["password-policy", "premium-user-policy"],
        "compliance_score": 0.95,
    },
)

Agent with subagents success

Sub-Agent Error​

await policy_checker.error(
    status_message=Exception("Policy verification failed"),
    stage="policy_check",
    metadata={
        "failed_policies": ["premium-user-policy"],
        "error_code": "POLICY_VIOLATION",
    },
)

Creating Deeper Hierarchies​

You can create multiple levels of sub-agents:

# Sub-sub-agent under policy checker
verifier = await policy_checker.add_agent({
    "name": "2FA Verifier",
    "input": {"user_id": "user-123"},
    "instructions": "You handle two-factor authentication verification processes.",
    "metadata": {
        "role": "two-factor-auth",
        "modelParameters": {
            "model": "gpt-3.5-turbo",
        },
    },
})

Agent with subagents-subagents

Complete the Agent and Trace​

Finally, complete your main agent and trace:

# Complete the main agent
await agent.success(
    output={
        "response": "Password reset link sent to user's email",
        "action_taken": "email-reset-link",
        "user_satisfied": True,
    },
    usage=TokenUsage(
        prompt_tokens=150,
        completion_tokens=85,
        total_tokens=235,
    ),
    metadata={
        "response_time": "2.1s",
        "confidence_score": 0.95,
    },
)

# Trace automatically completes when exiting context manager

Agent with success

Best Practices​

  1. Use context managers when possible - They provide automatic resource cleanup
  2. Always call await sdk.flush() before your application exits
  3. Use meaningful names for traces, agents, tools, and operations
  4. Include relevant metadata for debugging and analytics
  5. Track token usage in the usage field, not metadata
  6. Handle errors properly with descriptive error messages
  7. Use hierarchical agents for complex workflows
  8. Set appropriate tags for easy filtering and search
  9. Use type hints for better code maintainability
  10. Handle asyncio properly in your application lifecycle

Complete Examples​

import asyncio
import os
from voltagent import VoltAgentSDK, TokenUsage

async def context_manager_example():
    sdk = VoltAgentSDK(
        base_url=os.getenv("VOLTAGENT_BASE_URL", "https://api.voltagent.dev"),
        public_key=os.getenv("VOLTAGENT_PUBLIC_KEY"),
        secret_key=os.getenv("VOLTAGENT_SECRET_KEY"),
        auto_flush=True,
    )

    try:
        # Context manager automatically handles trace lifecycle
        async with sdk.trace(
            agentId="example-agent",
            input={"query": "Show me how to use the SDK"},
        ) as trace:

            # Add agent
            agent = await trace.add_agent({
                "name": "Example Agent",
                "input": {"task": "Demonstrate SDK usage"},
                "instructions": "You demonstrate how to use the VoltAgent SDK effectively.",
                "metadata": {
                    "modelParameters": {"model": "gpt-4"},
                },
            })

            # Add tool
            tool = await agent.add_tool({
                "name": "example-tool",
                "input": {"action": "demonstrate"},
            })

            await tool.success(
                output={"result": "Tool executed successfully"},
            )

            # Complete agent
            await agent.success(
                output={"response": "SDK demonstration completed"},
                usage=TokenUsage(prompt_tokens=50, completion_tokens=30, total_tokens=80),
            )

            # Trace automatically completes here

    except Exception as error:
        print(f"Example failed: {error}")
    finally:
        await sdk.shutdown()

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

Environment Variables​

For easier configuration, set these environment variables:

export VOLTAGENT_BASE_URL="https://api.voltagent.dev"
export VOLTAGENT_PUBLIC_KEY="your-public-key"
export VOLTAGENT_SECRET_KEY="your-secret-key"

Then initialize the SDK simply:

sdk = VoltAgentSDK(
    base_url=os.getenv("VOLTAGENT_BASE_URL"),
    public_key=os.getenv("VOLTAGENT_PUBLIC_KEY"),
    secret_key=os.getenv("VOLTAGENT_SECRET_KEY"),
    auto_flush=True,
)

Type Safety​

The Python SDK provides comprehensive type hints for better development experience:

from typing import Dict, Any, List, Optional
from voltagent import VoltAgentSDK, TokenUsage, TraceStatus

# SDK methods are fully typed
sdk: VoltAgentSDK = VoltAgentSDK(...)

# All data structures support type hints
metadata: Dict[str, Any] = {
    # Built-in SDK field for model configuration
    "modelParameters": {
        "model": "gpt-4",
        "temperature": 0.7,
    },
    # Custom user-defined fields (examples)
    "user_session_id": "session_123",
    "priority": "high",
    "environment": "production",
}

usage: TokenUsage = TokenUsage(
    prompt_tokens=100,
    completion_tokens=50,
    total_tokens=150,
)

Table of Contents