Python SDK

pip install agentbase-sdk

from agentbase_sdk import Agentbase

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

# Run an agent
response = agentbase.run_agent(
    message="Hello, analyze this data for me"
)

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

from agentbase_sdk import Agentbase

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

stream = agentbase.run_agent(
    message="Your task here",
    mode="fast",  # optional: "fast", "flash", "max"
    system="You are a helpful assistant",  # optional
    rules=["Be concise", "Show examples"],  # optional
    session="session-id",  # optional for conversation continuity
    streaming=True,  # optional, defaults to True
    mcp_servers=[  # optional
        {
            "serverName": "my-api",
            "serverUrl": "https://api.example.com/mcp"
        }
    ]
)

messages = agentbase.get_messages(session="session-id")

agentbase.clear_messages(session="session-id")

from agentbase_sdk import Agentbase, AgentbaseError

try:
    response = agentbase.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
from agentbase_sdk.types import AgentEvent, MCPServer

def run_agent(
    message: str,
    mode: Optional[str] = None,
    system: Optional[str] = None,
    rules: Optional[List[str]] = None,
    session: Optional[str] = None,
    streaming: bool = True,
    mcp_servers: Optional[List[MCPServer]] = None
) -> Iterator[AgentEvent]:
    pass

chat = agentbase.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 = agentbase.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 = agentbase.run_agent(
    message="Can you help me with async programming?",
    session=session_id
)

total_cost = 0

response = agentbase.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_sdk import AsyncAgentbase

async def main():
    agentbase = AsyncAgentbase(api_key="your-api-key")
    
    response = await agentbase.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_sdk import Agentbase

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

from agentbase_sdk import Agentbase
from agentbase_sdk.retry import RetryConfig

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