Standalone Usage Guide

August 14, 2026 · View on GitHub

This guide covers using the AI SDK directly without external frameworks like LangChain.

Prerequisites

Before using this SDK, you need:

An OpenMetadata or Collate instance with Dynamic Agents enabled
A Bot JWT token for API authentication

See Getting Your Credentials for detailed instructions on obtaining these.

Installation

Install the SDK with minimal dependencies:

pip install data-ai-sdk

Core dependencies are only httpx and pydantic - no framework lock-in.

Quick Start

from ai_sdk import AISdk

# Initialize client with explicit credentials
client = AISdk(
    host="https://your-org.getcollate.io",    # Your OpenMetadata/Collate URL
    token="eyJhbGciOiJSUzI1NiIs..."           # Your bot's JWT token
)

# Invoke an agent
response = client.agent("DataQualityPlannerAgent").call(
    "What data quality tests should I add for the customers table?"
)

print(response.response)

# Cleanup
client.close()

Client Configuration

Direct Initialization

from ai_sdk import AISdk

client = AISdk(
    host="https://metadata.example.com",
    token="your-bot-jwt-token",
    timeout=120.0,         # Request timeout in seconds
    verify_ssl=True,       # SSL certificate verification
    enable_async=False,    # Enable async operations
    max_retries=3,         # Retry attempts for transient errors
    retry_delay=1.0,       # Base delay between retries
)

From Environment Variables

Set your environment variables first:

# Required: Your OpenMetadata/Collate server URL
export AI_SDK_HOST="https://your-org.getcollate.io"

# Required: Your bot's JWT token (from Settings > Bots)
export AI_SDK_TOKEN="eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9..."

Then load them in Python:

from ai_sdk import AISdk, AISdkConfig

# Load from environment (reads AI_SDK_HOST and AI_SDK_TOKEN)
config = AISdkConfig.from_env()
client = AISdk.from_config(config)

All environment variables:

Variable	Required	Default	Description
`AI_SDK_HOST`	Yes	-	Your OpenMetadata/Collate server URL
`AI_SDK_TOKEN`	Yes	-	Bot JWT token (from Settings > Bots)
`AI_SDK_TIMEOUT`	No	`120`	Request timeout in seconds
`AI_SDK_VERIFY_SSL`	No	`true`	Verify SSL certificates (`true`/`false`)
`AI_SDK_DEBUG`	No	`false`	Enable debug logging
`AI_SDK_MAX_RETRIES`	No	`3`	Number of retry attempts for failed requests
`AI_SDK_RETRY_DELAY`	No	`1.0`	Base delay between retries (seconds)

With Overrides

# Start from environment, override specific values
config = AISdkConfig.from_env(
    timeout=30.0,
    enable_async=True,
)
client = AISdk.from_config(config)

Context Manager

with AISdk(host="...", token="...") as client:
    response = client.agent("MyAgent").call("Hello")
    print(response.response)
# Client automatically closed

Agent Invocation

Simple Call

agent = client.agent("DataQualityPlannerAgent")

response = agent.call("Analyze the customers table")

print(response.response)           # Agent's response text
print(response.conversation_id)    # ID for multi-turn conversations
print(response.tools_used)         # Tools the agent used
print(response.usage)              # Token usage statistics

With Parameters

response = agent.call(
    "Analyze this table",
    parameters={
        "table_name": "customers",
        "schema": "public",
    }
)

Multi-Turn Conversations

Manual Conversation ID

agent = client.agent("DataQualityPlannerAgent")

# First turn
response1 = agent.call("Analyze the orders table")

# Continue conversation
response2 = agent.call(
    "What specific tests would you recommend?",
    conversation_id=response1.conversation_id
)

# Third turn
response3 = agent.call(
    "Create those tests",
    conversation_id=response2.conversation_id
)

Using Conversation Helper

The Conversation class automatically manages conversation context:

from ai_sdk import AISdk, Conversation

client = AISdk(host="...", token="...")
agent = client.agent("DataQualityPlannerAgent")

# Create conversation
conv = Conversation(agent)

# Send messages - context is automatic
print(conv.send("Analyze the customers table"))
print(conv.send("Now create tests for the issues you found"))
print(conv.send("Show me the SQL for those tests"))

# Access conversation details
print(f"Turns: {len(conv)}")
print(f"Conversation ID: {conv.id}")
print(f"Tools used: {conv.tools_used}")

# Get history
for user_msg, assistant_msg in conv.history:
    print(f"User: {user_msg}")
    print(f"Assistant: {assistant_msg}")

# Start fresh
conv.reset()

Conversation Properties

Property	Type	Description
`id`	`str \| None`	Current conversation ID
`history`	`list[tuple[str, str]]`	(user, assistant) message pairs
`messages`	`list[dict]`	Chat format with `role` and `content`
`responses`	`list[InvokeResponse]`	Raw response objects
`tools_used`	`list[str]`	All tools used across turns

Streaming Responses

Get real-time output as the agent generates:

agent = client.agent("SqlQueryAgent")

for event in agent.stream("Generate SQL to find duplicate records"):
    match event.type:
        case "start":
            print("Agent started...")
        case "content":
            print(event.content, end="", flush=True)
        case "tool_use":
            print(f"\n[Using tool: {event.tool_name}]")
        case "end":
            print("\nDone!")
        case "error":
            print(f"\nError: {event.error}")

Stream Event Types

Type	Fields	Description
`start`	`conversation_id`	Agent started processing
`content`	`content`	Text chunk from agent
`tool_use`	`tool_name`	Agent is using a tool
`end`	-	Agent finished
`error`	`error`	Error occurred

Streaming with Conversation

conv = Conversation(agent)

# Stream first turn
for event in conv.stream("Analyze this table"):
    if event.type == "content":
        print(event.content, end="")

# Note: Streaming doesn't auto-update history
# Use send() for tracked multi-turn conversations

Listing Agents

Discover available API-enabled agents:

agents = client.agents.list(limit=20)

for agent in agents:
    print(f"Name: {agent.name}")
    print(f"Display: {agent.display_name}")
    print(f"Description: {agent.description}")
    print(f"Skills: {agent.skills}")
    print(f"API Enabled: {agent.api_enabled}")
    print()

Get Single Agent Info

agent = client.agent("DataQualityPlannerAgent")
info = agent.get_info()

print(info.name)
print(info.description)
print(info.skills)

Creating Agents

Create new dynamic agents programmatically:

from ai_sdk.models import CreateAgentRequest

# Create a simple agent
agent = client.agents.create(CreateAgentRequest(
    name="MyDataAgent",
    description="An agent for data analysis tasks",
    persona="DataAnalyst",  # Name of an existing persona
    api_enabled=True,
))
print(f"Created agent: {agent.name}")

# Create an agent with full configuration
agent = client.agents.create(CreateAgentRequest(
    name="AdvancedAgent",
    description="An advanced agent with custom configuration",
    persona="DataAnalyst",
    display_name="Advanced Data Agent",
    api_enabled=True,
    skills=["search", "query", "analyze"],
    prompt="Analyze user data and provide insights",
    provider="openai",
    bot_name="my-bot",  # Bot for executing actions
))

CreateAgentRequest Fields

Field	Type	Required	Description
`name`	`str`	Yes	Unique identifier for the agent
`description`	`str`	Yes	Agent description
`persona`	`str`	Yes	Name of the persona to use
`display_name`	`str`	No	Human-readable name
`api_enabled`	`bool`	No	Enable API access (default: False)
`skills`	`list[str]`	No	List of skill names
`prompt`	`str`	No	Default task/prompt
`provider`	`str`	No	LLM provider
`bot_name`	`str`	No	Bot for executing actions
`mode`	`str`	No	Agent mode
`icon`	`str`	No	Icon URL or name
`knowledge`	`list[str]`	No	Knowledge sources
`schedule`	`str`	No	Cron schedule

Bot Operations

Bots are service accounts used for API authentication and actions.

# List all bots
bots = client.bots.list(limit=20)
for bot in bots:
    print(f"{bot.name}: {bot.display_name}")

# Get a specific bot
bot = client.bots.get("my-bot-name")
print(f"Bot: {bot.name}")
print(f"Display Name: {bot.display_name}")

BotInfo Fields

Field	Type	Description
`name`	`str`	Bot identifier
`display_name`	`str`	Human-readable name
`description`	`str`	Bot description

Bot Errors

from ai_sdk.exceptions import BotNotFoundError

try:
    bot = client.bots.get("nonexistent-bot")
except BotNotFoundError as e:
    print(f"Bot not found: {e.bot_name}")

Persona Operations

Personas define the behavior and personality of agents.

from ai_sdk.models import CreatePersonaRequest

# List all personas
personas = client.personas.list(limit=20)
for persona in personas:
    print(f"{persona.name}: {persona.description}")

# Get a specific persona
persona = client.personas.get("DataAnalyst")
print(f"Persona: {persona.name}")
print(f"Prompt: {persona.prompt[:100]}...")

# Create a new persona
new_persona = client.personas.create(CreatePersonaRequest(
    name="CustomAnalyst",
    description="A specialized analyst for custom domains",
    prompt="You are an expert analyst who helps users understand complex data...",
))
print(f"Created persona: {new_persona.name}")

CreatePersonaRequest Fields

Field	Type	Required	Description
`name`	`str`	Yes	Unique identifier
`description`	`str`	Yes	Persona description
`prompt`	`str`	Yes	System prompt defining behavior
`display_name`	`str`	No	Human-readable name
`provider`	`str`	No	LLM provider

PersonaInfo Fields

Field	Type	Description
`name`	`str`	Persona identifier
`display_name`	`str`	Human-readable name
`description`	`str`	Persona description
`prompt`	`str`	System prompt

Persona Errors

from ai_sdk.exceptions import PersonaNotFoundError

try:
    persona = client.personas.get("nonexistent")
except PersonaNotFoundError as e:
    print(f"Persona not found: {e.persona_name}")

Skill Operations

Skills are capabilities that can be assigned to agents.

# List all skills
skills = client.skills.list(limit=50)
for skill in skills:
    print(f"{skill.name}: {skill.description}")

# Get a specific skill
skill = client.skills.get("search")
print(f"Skill: {skill.name}")
print(f"Description: {skill.description}")

SkillInfo Fields

Field	Type	Description
`name`	`str`	Skill identifier
`display_name`	`str`	Human-readable name
`description`	`str`	Skill description

Skill Errors

from ai_sdk.exceptions import SkillNotFoundError

try:
    skill = client.skills.get("nonexistent")
except SkillNotFoundError as e:
    print(f"Skill not found: {e.skill_name}")

Async Operations

Enable async for concurrent operations:

import asyncio
from ai_sdk import AISdk

async def main():
    client = AISdk(
        host="https://metadata.example.com",
        token="your-token",
        enable_async=True  # Required
    )

    agent = client.agent("DataQualityPlannerAgent")

    # Async call
    response = await agent.acall("Analyze the orders table")
    print(response.response)

    # Async streaming
    async for event in await agent.astream("Generate SQL query"):
        if event.type == "content":
            print(event.content, end="")

    # Async list
    agents = await client.agents.alist()

    # Cleanup
    await client.aclose()
    client.close()

asyncio.run(main())

Async Conversation

from ai_sdk import Conversation

async def chat():
    client = AISdk(host="...", token="...", enable_async=True)
    conv = Conversation(client.agent("MyAgent"))

    # Async send
    response1 = await conv.asend("First message")
    response2 = await conv.asend("Follow up")

    # Async stream
    async for event in await conv.astream("Stream this"):
        print(event.content, end="")

Async Context Manager

async with AISdk(host="...", token="...", enable_async=True) as client:
    response = await client.agent("MyAgent").acall("Hello")

Error Handling

Handle specific error types:

from ai_sdk import AISdk
from ai_sdk.exceptions import (
    AISdkError,
    AuthenticationError,
    AgentNotFoundError,
    AgentNotEnabledError,
    RateLimitError,
    AgentExecutionError,
)

client = AISdk(host="...", token="...")

try:
    response = client.agent("MyAgent").call("Hello")
    print(response.response)

except AuthenticationError:
    print("Invalid or expired token")
    print("Check your AI_SDK_TOKEN")

except AgentNotFoundError as e:
    print(f"Agent not found: {e.agent_name}")
    print("Verify the agent exists in your Metadata instance")

except AgentNotEnabledError as e:
    print(f"Agent '{e.agent_name}' is not API-enabled")
    print("Enable API access in AI Studio")

except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after} seconds")
    if e.retry_after:
        time.sleep(e.retry_after)

except AgentExecutionError as e:
    print(f"Agent execution failed: {e.message}")

except AISdkError as e:
    print(f"Metadata error ({e.status_code}): {e.message}")

Exception Hierarchy

AISdkError (base)
├── AuthenticationError (401)
├── AgentNotFoundError (404)
├── AgentNotEnabledError (403)
├── BotNotFoundError (404)
├── PersonaNotFoundError (404)
├── SkillNotFoundError (404)
├── RateLimitError (429)
└── AgentExecutionError (500)

Debug Logging

Enable verbose logging for debugging:

from ai_sdk import AISdkConfig, AISdk
from ai_sdk._logging import set_debug

# Option 1: Via config
config = AISdkConfig.from_env(debug=True)
client = AISdk.from_config(config)

# Option 2: Direct toggle
set_debug(True)

Custom Logging Configuration

from ai_sdk._logging import configure_logging
import logging

# Custom format
configure_logging(
    level=logging.DEBUG,
    format_string="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

# Or use your own handler
handler = logging.FileHandler("metadata.log")
configure_logging(handler=handler)

Testing with Mocks

Use the ABC interfaces for testing:

from ai_sdk.protocols import AgentProtocol
from ai_sdk.models import InvokeResponse

class MockAgent(AgentProtocol):
    """Mock agent for testing."""

    def __init__(self, responses: list[str]):
        self._responses = iter(responses)

    @property
    def name(self) -> str:
        return "MockAgent"

    def call(self, message, **kwargs) -> InvokeResponse:
        return InvokeResponse(
            conversation_id="mock-123",
            response=next(self._responses),
            tools_used=[],
        )

    # Implement other abstract methods...

# Use in tests
def test_my_function():
    mock = MockAgent(["Response 1", "Response 2"])
    result = my_function(mock)
    assert result == expected

Data Models

InvokeResponse

response = agent.call("...")

response.conversation_id  # str - For multi-turn
response.response         # str - Agent's response
response.tools_used       # list[str] - Tools used
response.usage            # Usage | None - Token stats

Usage

if response.usage:
    print(response.usage.prompt_tokens)
    print(response.usage.completion_tokens)
    print(response.usage.total_tokens)

StreamEvent

event.type            # str - Event type
event.content         # str | None - Text content
event.tool_name       # str | None - Tool being used
event.conversation_id # str | None - Conversation ID
event.error           # str | None - Error message

AgentInfo

info = agent.get_info()

info.name          # str - Agent identifier
info.display_name  # str - Human-readable name
info.description   # str - Agent description
info.skills     # list[str] - Capabilities
info.api_enabled   # bool - API access enabled

Complete Example

#!/usr/bin/env python3
"""Complete standalone SDK usage example."""

import sys
from ai_sdk import AISdk, AISdkConfig, Conversation
from ai_sdk.models import CreatePersonaRequest, CreateAgentRequest
from ai_sdk.exceptions import AISdkError

def main():
    # Load configuration
    try:
        config = AISdkConfig.from_env()
    except ValueError as e:
        print(f"Configuration error: {e}")
        print("Set AI_SDK_HOST and AI_SDK_TOKEN environment variables")
        sys.exit(1)

    # Create client
    client = AISdk.from_config(config)

    try:
        # List available agents
        print("Available agents:")
        for agent in client.agents.list():
            print(f"  - {agent.name}: {agent.description[:50]}...")

        # List bots
        print("\n--- Bots ---")
        for bot in client.bots.list():
            print(f"  - {bot.name}: {bot.display_name}")

        # List personas
        print("\n--- Personas ---")
        for persona in client.personas.list():
            print(f"  - {persona.name}: {persona.description[:50]}...")

        # List skills
        print("\n--- Skills ---")
        for skill in client.skills.list():
            print(f"  - {skill.name}: {skill.description[:50]}...")

        # Simple invocation
        print("\n--- Simple Invocation ---")
        agent = client.agent("DataQualityPlannerAgent")
        response = agent.call("What should I test for a customers table?")
        print(response.response[:500])

        # Multi-turn conversation
        print("\n--- Multi-turn Conversation ---")
        conv = Conversation(agent)
        print(conv.send("Analyze the orders table"))
        print(conv.send("Create tests for the top issue"))
        print(f"\nConversation had {len(conv)} turns")

        # Streaming
        print("\n--- Streaming ---")
        for event in agent.stream("Generate a simple SQL query"):
            if event.type == "content" and event.content:
                print(event.content, end="", flush=True)
        print()

        # Create a persona (uncomment to run)
        # print("\n--- Create Persona ---")
        # persona = client.personas.create(CreatePersonaRequest(
        #     name="MyCustomPersona",
        #     description="A custom persona for testing",
        #     prompt="You are a helpful assistant..."
        # ))
        # print(f"Created persona: {persona.name}")

        # Create an agent (uncomment to run)
        # print("\n--- Create Agent ---")
        # new_agent = client.agents.create(CreateAgentRequest(
        #     name="MyCustomAgent",
        #     description="A custom agent for testing",
        #     persona="DataAnalyst",
        #     api_enabled=True,
        # ))
        # print(f"Created agent: {new_agent.name}")

    except AISdkError as e:
        print(f"Error: {e}")
        sys.exit(1)

    finally:
        client.close()

if __name__ == "__main__":
    main()

Best Practices

Use context managers - Ensure proper cleanup
Handle specific exceptions - Don't catch generic Exception
Use Conversation for multi-turn - Avoid manual ID tracking
Enable async for concurrency - Better throughput
Configure logging in production - Integrate with your logging
Use environment config - Keep secrets out of code

API Reference

See the inline documentation for complete API details:

from ai_sdk import AISdk
help(AISdk)

from ai_sdk import Conversation
help(Conversation)