Overview

The synchronous client provides blocking operations for interacting with RunAgent. Perfect for scripts, simple applications, and when you don’t need concurrent operations.

Initialization

from runagent import RunAgentClient

# Basic initialization
client = RunAgentClient(agent_id="your-agent-id")

# With all options
client = RunAgentClient(
    agent_id="your-agent-id",
    api_key="your-api-key",
    base_url="https://api.run-agent.ai",
    timeout=30,
    max_retries=3
)

Basic Usage

Simple Request

result = client.run_generic({
    "query": "What's the weather like?",
    "location": "San Francisco"
})

print(result)
# {'weather': 'sunny', 'temperature': 72, 'unit': 'fahrenheit'}

With Error Handling

try:
    result = client.run_generic({"query": "Hello"})
    print(f"Success: {result}")
except RunAgentError as e:
    print(f"Error: {e.message}")
    print(f"Code: {e.code}")
except Exception as e:
    print(f"Unexpected error: {e}")

Request Options

Timeout Control

# Short timeout for quick operations
result = client.run_generic(
    {"query": "Quick question"},
    timeout=5  # 5 seconds
)

# Longer timeout for complex operations
result = client.run_generic(
    {"query": "Analyze this dataset", "data": large_data},
    timeout=120  # 2 minutes
)

Custom Headers

result = client.run_generic(
    {"query": "Test"},
    headers={
        "X-Request-ID": "unique-123",
        "X-User-ID": "user-456"
    }
)

Response Handling

Accessing Response Metadata

response = client.run_generic_with_metadata({"query": "Test"})

# Access result and metadata
print(response.result)  # Actual agent response
print(response.execution_time)  # Time taken
print(response.tokens_used)  # Token usage
print(response.agent_version)  # Agent version

Parsing Complex Responses

result = client.run_generic({
    "action": "analyze",
    "data": {"sales": [100, 200, 300]}
})

# Handle structured response
if result.get("status") == "success":
    analysis = result["analysis"]
    print(f"Total: {analysis['total']}")
    print(f"Average: {analysis['average']}")

Advanced Features

Batch Processing

# Process multiple requests
requests = [
    {"query": "Question 1"},
    {"query": "Question 2"},
    {"query": "Question 3"}
]

results = []
for request in requests:
    try:
        result = client.run_generic(request)
        results.append({"input": request, "output": result})
    except Exception as e:
        results.append({"input": request, "error": str(e)})

Session Management

# Maintain conversation context
session_id = "conv-123"

# First message
response1 = client.run_generic({
    "query": "My name is Alice",
    "session_id": session_id
})

# Follow-up message
response2 = client.run_generic({
    "query": "What's my name?",
    "session_id": session_id
})
# Should remember "Alice"

Retry Configuration

# Custom retry logic
client = RunAgentClient(
    agent_id="your-agent-id",
    max_retries=5,
    retry_delay=1.0,  # Start with 1 second
    retry_backoff=2.0  # Double each retry
)

Best Practices

  1. Reuse Client Instances

    # Good - create once
    client = RunAgentClient(agent_id="...")
    for item in items:
        client.run_generic(item)
    
    # Bad - create every time
    for item in items:
        client = RunAgentClient(agent_id="...")
        client.run_generic(item)
    
  2. Handle Timeouts

    try:
        result = client.run_generic(data, timeout=10)
    except TimeoutError:
        # Handle timeout appropriately
        result = {"error": "Request timed out"}
    
  3. Log Requests

    import logging
    
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)
    
    def run_with_logging(client, data):
        logger.info(f"Request: {data}")
        try:
            result = client.run_generic(data)
            logger.info(f"Response: {result}")
            return result
        except Exception as e:
            logger.error(f"Error: {e}")
            raise
    

See Also