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
)
result = client.run_generic(
{"query": "Test"},
headers={
"X-Request-ID": "unique-123",
"X-User-ID": "user-456"
}
)
Response Handling
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
-
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)
-
Handle Timeouts
try:
result = client.run_generic(data, timeout=10)
except TimeoutError:
# Handle timeout appropriately
result = {"error": "Request timed out"}
-
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