Python SDK API Reference

RunAgentClient(
    agent_id: str,
    api_key: Optional[str] = None,
    base_url: str = "https://api.run-agent.ai",
    timeout: int = 30,
    max_retries: int = 3,
    headers: Optional[Dict[str, str]] = None
)

run_generic(
    input_data: Dict[str, Any],
    timeout: Optional[int] = None,
    headers: Optional[Dict[str, str]] = None
) -> Dict[str, Any]

run_generic_stream(
    input_data: Dict[str, Any],
    timeout: Optional[int] = None,
    headers: Optional[Dict[str, str]] = None
) -> Generator[str, None, None]

AsyncRunAgentClient(
    agent_id: str,
    api_key: Optional[str] = None,
    base_url: str = "https://api.run-agent.ai",
    timeout: int = 30,
    max_retries: int = 3,
    headers: Optional[Dict[str, str]] = None
)

async run_generic(
    input_data: Dict[str, Any],
    timeout: Optional[int] = None,
    headers: Optional[Dict[str, str]] = None
) -> Dict[str, Any]

async run_generic_stream(
    input_data: Dict[str, Any],
    timeout: Optional[int] = None,
    headers: Optional[Dict[str, str]] = None
) -> AsyncGenerator[str, None]

class RunAgentError(Exception):
    def __init__(self, message: str, code: Optional[str] = None, details: Optional[Dict] = None):
        self.message = message
        self.code = code
        self.details = details

from typing import Dict, Any, Optional, Generator, AsyncGenerator

# Input/Output types
InputData = Dict[str, Any]
OutputData = Dict[str, Any]

# Stream types
StreamChunk = str
StreamGenerator = Generator[StreamChunk, None, None]
AsyncStreamGenerator = AsyncGenerator[StreamChunk, None]

{
  "api_key": "your-api-key",
  "api_url": "https://api.run-agent.ai",
  "timeout": 30,
  "retry_config": {
    "max_retries": 3,
    "retry_delay": 1.0,
    "retry_backoff": 2.0
  }
}

from runagent import RunAgentClient, RunAgentError
import asyncio

# Synchronous example
def sync_example():
    client = RunAgentClient(agent_id="agent-123")
    
    try:
        # Simple request
        result = client.run_generic({"query": "Hello"})
        print(f"Response: {result}")
        
        # Streaming
        for chunk in client.run_generic_stream({"query": "Tell a story"}):
            print(chunk, end="")
            
    except RunAgentError as e:
        print(f"Error: {e.message}")

# Asynchronous example
async def async_example():
    from runagent import AsyncRunAgentClient
    
    client = AsyncRunAgentClient(agent_id="agent-123")
    
    # Concurrent requests
    tasks = [
        client.run_generic({"query": f"Question {i}"})
        for i in range(5)
    ]
    results = await asyncio.gather(*tasks)
    
    # Streaming
    async for chunk in client.run_generic_stream({"query": "Stream test"}):
        print(chunk, end="")

# Run examples
sync_example()
asyncio.run(async_example())