Python SDK

pip install yourautomation

poetry add yourautomation

import os from yourautomation import YourAutomation client = YourAutomation( api_key=os.environ["YA_API_KEY"], base_url="https://api.yourautomation.ai/v1", # default timeout=30.0, # seconds max_retries=3, )

from yourautomation import AsyncYourAutomation client = AsyncYourAutomation( api_key=os.environ["YA_API_KEY"], ) # Use with async/await agent = await client.agents.create( name="Support Agent", model="claude-sonnet-4-20250514", system_prompt="You are helpful...", )

# Create agent = client.agents.create( name="Support Agent", model="claude-sonnet-4-20250514", system_prompt="You are a helpful customer support agent...", tools=[ { "name": "lookup_order", "description": "Look up an order by ID", "parameters": { "type": "object", "properties": { "order_id": {"type": "string"} }, "required": ["order_id"], }, "handler": "https://api.example.com/webhooks/orders", } ], parameters={"temperature": 0.3, "max_tokens": 1024}, ) # List agents = client.agents.list(limit=20) for a in agents.data: print(a.id, a.name, a.status) # Retrieve agent = client.agents.get("agt_abc123") # Update updated = client.agents.update("agt_abc123", system_prompt="Updated instructions...", ) # Delete client.agents.delete("agt_abc123")

# Standard execution result = client.agents.execute( "agt_abc123", input="What is your return policy?", conversation_id="conv_optional", ) print(result.output) # Agent's response print(result.tool_calls) # Tools called print(result.usage) # Token usage # With session memory result = client.agents.execute( "agt_abc123", input="Check my order status", memory={"session": {"user_name": "Sarah", "account_id": "ACC-789"}}, )

# Sync streaming stream = client.agents.execute( "agt_abc123", input="Explain your pricing", stream=True, ) for event in stream: if event.type == "text_delta": print(event.text, end="", flush=True) elif event.type == "tool_call_start": print(f"\nCalling tool: {event.name}") elif event.type == "done": print(f"\nTokens: {event.usage.total_tokens}") # Async streaming async for event in await async_client.agents.execute( "agt_abc123", input="Explain your pricing", stream=True, ): if event.type == "text_delta": print(event.text, end="", flush=True)

from yourautomation.types import ( Agent, Execution, StreamEvent, Tool, ToolCall, Usage, Conversation, ) # All responses are typed Pydantic models agent: Agent = client.agents.get("agt_abc123") print(agent.name) # str print(agent.created_at) # datetime print(agent.parameters) # AgentParameters result: Execution = client.agents.execute(agent.id, input="Hello") print(result.usage.total_tokens) # int

from yourautomation import APIError, RateLimitError, AuthenticationError try: result = client.agents.execute("agt_abc123", input="Hello") except RateLimitError as e: print(f"Rate limited. Retry after {e.retry_after}s") time.sleep(e.retry_after) except AuthenticationError as e: print(f"Auth error: {e.message}") except APIError as e: print(f"Error {e.status_code}: {e.code} - {e.message}") print(f"Request ID: {e.request_id}")

# Automatically closes connections with YourAutomation(api_key=os.environ["YA_API_KEY"]) as client: result = client.agents.execute("agt_abc123", input="Hello") print(result.output) # Async context manager async with AsyncYourAutomation(api_key=os.environ["YA_API_KEY"]) as client: result = await client.agents.execute("agt_abc123", input="Hello") print(result.output)

from fastapi import FastAPI from fastapi.responses import StreamingResponse from yourautomation import AsyncYourAutomation app = FastAPI() client = AsyncYourAutomation(api_key=os.environ["YA_API_KEY"]) @app.post("/chat") async def chat(input: str, agent_id: str): stream = await client.agents.execute( agent_id, input=input, stream=True ) async def generate(): async for event in stream: if event.type == "text_delta": yield f"data: {event.model_dump_json()}\n\n" elif event.type == "done": yield f"data: {event.model_dump_json()}\n\n" return StreamingResponse(generate(), media_type="text/event-stream")