Python SDK

pip install agentbase-sdk

from agentbase import Agentbase

client = Agentbase(api_key="your-api-key")

# Run an agent (root-level method)
response = client.run_agent(
    message="Hello, analyze this data for me"
)

# Or use the agent resource
response2 = client.agent.run(
    message="Hello, analyze this data for me"
)

# Handle streaming response
for event in response:
    print(event)

from agentbase import Agentbase

client = Agentbase(
    api_key="your-api-key",
    base_url="https://api.agentbase.sh",  # optional
    timeout=30  # optional, in seconds
)

stream = client.run_agent(
    message="Your task here",  # required
    session="session-id",  # optional: continue conversation
    mode="base",  # optional: "base" | "flash" | "max", defaults to "base"
    system="You are a helpful assistant",  # optional: system prompt
    rules=["Be concise", "Show examples"],  # optional: array of constraints
    workflows=[  # optional: declarative workflows
        {
            "id": "workflow_1",
            "name": "example_workflow",
            "description": "Example workflow description",
            "steps": [
                {
                    "id": "step_1",
                    "name": "first_step",
                    "description": "What this step should accomplish",
                    "depends_on": [],
                    "optional": False,  # optional
                    "retry_policy": {  # optional
                        "max_attempts": 3,
                        "backoff": "exponential"
                    },
                    "output_schema": {  # optional
                        "type": "object",
                        "properties": {
                            "result": {"type": "string"}
                        }
                    }
                }
            ]
        }
    ],
    mcp_servers=[  # optional: MCP server configs
        {
            "serverName": "my-api",
            "serverUrl": "https://api.example.com/mcp"
        }
    ],
    background=False,  # optional: run agent asynchronously, defaults to False
    callback={  # optional: use with background=True
        "url": "https://your-server.com/webhook",
        "headers": {
            "Authorization": "Bearer your-token"
        }
    },
    datastores=[  # optional: data sources for the agent
        {
            "id": "ds_1234567890abcdef",
            "name": "my-datastore"
        }
    ],
    queries=[  # optional: custom datastore queries
        {
            "name": "getUserById",
            "description": "Fetch user details by their ID",
            "query": "SELECT * FROM users WHERE id = ?"
        }
    ],
    streaming_tokens=False,  # optional: stream tokens, defaults to False
    final_output={  # optional: structured final output
        "name": "task_summary",
        "strict": True,
        "schema": {
            "type": "object",
            "properties": {
                "summary": {"type": "string"},
                "outcome": {
                    "type": "string",
                    "enum": ["success", "partial_success", "failure"]
                }
            },
            "required": ["summary", "outcome"]
        }
    }
)

stream = client.agent.run(
    message="Your task here",  # required
    session="session-id",  # optional
    mode="base",  # optional
    system="You are a helpful assistant",  # optional
    rules=["Be concise", "Show examples"],  # optional
    workflows=[...],  # optional
    mcp_servers=[...],  # optional
    background=False,  # optional
    callback={...},  # optional
    datastores=[...],  # optional
    queries=[...],  # optional
    streaming_tokens=False,  # optional
    final_output={...}  # optional
)

messages = client.messages.get(session="session-id")

result = client.messages.clear(session="session-id")

from agentbase import Agentbase, AgentbaseError

try:
    response = client.run_agent(message="Your task")
except AgentbaseError as e:
    print(f"API Error: {e.message}")
    print(f"Status: {e.status}")
except Exception as e:
    print(f"Unexpected error: {e}")

from typing import List, Dict, Optional, Iterator, TypedDict, Any

class WorkflowStep(TypedDict, total=False):
    id: str
    name: str
    description: str
    depends_on: List[str]
    optional: bool
    retry_policy: Dict[str, Any]
    output_schema: Dict[str, Any]

class Workflow(TypedDict):
    id: str
    name: str
    description: str
    steps: List[WorkflowStep]

class MCPServer(TypedDict):
    serverName: str
    serverUrl: str

class CallbackConfig(TypedDict, total=False):
    url: str
    headers: Dict[str, str]

class Datastore(TypedDict):
    id: str
    name: str

class Query(TypedDict):
    name: str
    description: str
    query: str

class FinalOutputConfig(TypedDict):
    name: str
    strict: bool
    schema: Dict[str, Any]

class AgentEvent(TypedDict, total=False):
    type: str
    content: str
    session: str
    cost: str
    balance: float

def run_agent(
    message: str,
    session: Optional[str] = None,
    mode: Optional[str] = None,
    system: Optional[str] = None,
    rules: Optional[List[str]] = None,
    workflows: Optional[List[Workflow]] = None,
    mcp_servers: Optional[List[MCPServer]] = None,
    background: bool = False,
    callback: Optional[CallbackConfig] = None,
    datastores: Optional[List[Datastore]] = None,
    queries: Optional[List[Query]] = None,
    streaming_tokens: bool = False,
    final_output: Optional[FinalOutputConfig] = None
) -> Iterator[AgentEvent]:
    pass

chat = client.run_agent(
    message="What's the weather like today?"
)

for event in chat:
    if event["type"] == "agent_message":
        print(f"Agent: {event['content']}")

session_id = None

# First message
response1 = client.run_agent(
    message="Hello, I'm working on a Python project"
)

for event in response1:
    if "session" in event:
        session_id = event["session"]

# Continue conversation
response2 = client.agent.run(
    message="Can you help me with async programming?",
    session=session_id
)

total_cost = 0

response = client.run_agent(
    message="Analyze this dataset"
)

for event in response:
    if event["type"] == "agent_cost":
        cost = float(event.get("cost", 0))
        total_cost += cost
        print(f"Current cost: ${event['cost']}")
        print(f"Total spent: ${total_cost}")

import asyncio
from agentbase import AsyncAgentbase

async def main():
    client = AsyncAgentbase(api_key="your-api-key")

    response = await client.run_agent(
        message="Analyze this data asynchronously"
    )

    async for event in response:
        print(event)

# Run the async function
asyncio.run(main())

with Agentbase(api_key="your-api-key") as client:
    response = client.run_agent(message="Your task")
    for event in response:
        print(event)

import httpx
from agentbase import Agentbase

# Use custom HTTP client
custom_http_client = httpx.Client(timeout=60)
client = Agentbase(
    api_key="your-api-key",
    http_client=custom_http_client
)

from agentbase import Agentbase
from agentbase.retry import RetryConfig

client = Agentbase(
    api_key="your-api-key",
    retry_config=RetryConfig(
        max_retries=3,
        backoff_factor=1.5
    )
)