Python SDK

Interact with Asteroid Agents using Python

1. Installation

   pip install --upgrade asteroid-odyssey
   

   Or with uv:

   uv pip install asteroid-odyssey
   
2. Create your Python file (run_agent.py)

   from asteroid_odyssey import (
       Configuration,
       ApiClient,
       AgentsApi,
       ExecutionApi,
       AgentsAgentExecuteAgentRequest,
   )
   import time
   
   # Configure the client
   config = Configuration(
       api_key={"ApiKeyAuth": "your-api-key"}
   )
   client = ApiClient(config)
   
   # Execute an agent
   agents_api = AgentsApi(client)
   request = AgentsAgentExecuteAgentRequest(
       inputs={"username": "[e-mail]"}
   )
   response = agents_api.agent_execute_post(
       agent_id="your-agent-id",
       agents_agent_execute_agent_request=request
   )
   execution_id = response.execution_id
   print(f"Execution started: {execution_id}")
   
   # Wait for result
   execution_api = ExecutionApi(client)
   while True:
       result = execution_api.execution_get(execution_id=execution_id)
       status = result.status.value
       print(f"Status: {status}")
       if status in ["completed", "failed", "cancelled"]:
           break
       time.sleep(2)
   
   print(f"Final status: {status}")
   
3. Run the agent

   Replace your API key, your agent_id, and update the inputs with your agent’s variables (or remove if no variables), then run:

   python run_agent.py
   

Agent Creation: Agents can only be created through the Asteroid Platform web interface. The API is designed for executing existing agents, not creating them.

API Classes

The SDK provides four main API classes:

Class	Purpose
AgentsApi	List agents, execute agents
ExecutionApi	Get execution status, activities, send messages
FilesApi	Upload files, stage temporary files
AgentProfilesApi	Manage agent profiles (browser configs, proxies, cookies)

Quick Reference

AgentsApi

from asteroid_odyssey import AgentsApi, AgentsAgentExecuteAgentRequest

agents_api = AgentsApi(client)

# List all agents
response = agents_api.agent_list()
for agent in response.items:
    print(f"{agent.id}: {agent.name}")

# List with pagination and search
response = agents_api.agent_list(
    page_size=10,
    page=1,
    search_name="My Agent"
)

# Execute an agent
request = AgentsAgentExecuteAgentRequest(
    inputs={"username": "[e-mail]", "date": "2025-01-15"},
    agent_profile_id="optional-profile-id"  # Optional
)
response = agents_api.agent_execute_post(
    agent_id="your-agent-id",
    agents_agent_execute_agent_request=request
)
execution_id = response.execution_id

ExecutionApi

from asteroid_odyssey import ExecutionApi, AgentsExecutionUserMessagesAddTextBody

execution_api = ExecutionApi(client)

# List executions
response = execution_api.executions_list(page_size=10)
for execution in response.items:
    print(f"{execution.id}: {execution.status.value}")

# Get execution details
execution = execution_api.execution_get(execution_id="your-execution-id")
print(f"Status: {execution.status.value}")
print(f"Agent ID: {execution.agent_id}")

# Get execution activities
activities = execution_api.execution_activities_get(execution_id="your-execution-id")
for activity in activities:
    print(f"Activity: {activity.payload}")

# Send a message to a running execution
body = AgentsExecutionUserMessagesAddTextBody(text="Hello from the user!")
execution_api.execution_user_messages_add(
    execution_id="your-execution-id",
    agents_execution_user_messages_add_text_body=body
)

FilesApi

from asteroid_odyssey import FilesApi

files_api = FilesApi(client)

# Get files from an execution
files = files_api.execution_context_files_get(execution_id="your-execution-id")
for f in files:
    print(f"{f.name}: {f.size} bytes")

# Stage temporary files (for use in execution)
file_content = b"Hello World!"
response = files_api.temp_files_stage(
    organization_id="your-org-id",
    files=[("document.txt", file_content)]
)
print(f"Staged files: {response.files}")

# Upload files to a running execution
with open("myfile.pdf", "rb") as f:
    files_api.execution_context_files_upload(
        execution_id="your-execution-id",
        files=[("myfile.pdf", f.read())]
    )

AgentProfilesApi

from asteroid_odyssey import (
    AgentProfilesApi,
    AgentsProfileCreateAgentProfileRequest,
    AgentsProfileUpdateAgentProfileRequest,
    AgentsProfileCountryCode,
    AgentsProfileProxyType,
)

profiles_api = AgentProfilesApi(client)

# List profiles for an organization
response = profiles_api.agent_profiles_list(organization_id="your-org-id")
for profile in response.items:
    print(f"{profile.id}: {profile.name}")

# Get a specific profile
profile = profiles_api.agent_profile_get(profile_id="your-profile-id")
print(f"Name: {profile.name}")
print(f"OS: {profile.operating_system}")

# Create a new profile
request = AgentsProfileCreateAgentProfileRequest(
    name="My Profile",
    organization_id="your-org-id",
    country_code=AgentsProfileCountryCode.US,
    proxy_type=AgentsProfileProxyType.RESIDENTIAL,
)
new_profile = profiles_api.agent_profiles_create(
    agents_profile_create_agent_profile_request=request
)

# Update a profile
update_request = AgentsProfileUpdateAgentProfileRequest(
    name="Updated Profile Name"
)
profiles_api.agent_profile_update(
    profile_id="your-profile-id",
    agents_profile_update_agent_profile_request=update_request
)

# Delete a profile
profiles_api.agent_profile_delete(profile_id="your-profile-id")

# Clear browser cache for a profile
profiles_api.agent_profile_clear_browser_cache(profile_id="your-profile-id")

Complete Examples

Execute an agent and wait for completion

from asteroid_odyssey import (
    Configuration,
    ApiClient,
    AgentsApi,
    ExecutionApi,
    AgentsAgentExecuteAgentRequest,
)
import time
import os

# Terminal statuses
TERMINAL_STATUSES = {"completed", "failed", "cancelled"}

def main():
    # Configure client
    config = Configuration(
        api_key={"ApiKeyAuth": os.environ.get("ASTEROID_API_KEY")}
    )
    client = ApiClient(config)

    # Execute agent
    agents_api = AgentsApi(client)
    request = AgentsAgentExecuteAgentRequest(
        inputs={
            "username": "[e-mail]",
            "order_id": "ORD-12345"
        }
    )
    response = agents_api.agent_execute_post(
        agent_id="your-agent-id",
        agents_agent_execute_agent_request=request
    )
    execution_id = response.execution_id
    print(f"Execution started: {execution_id}")

    # Poll for completion
    execution_api = ExecutionApi(client)
    while True:
        result = execution_api.execution_get(execution_id=execution_id)
        status = result.status.value
        print(f"Status: {status}")

        if status in TERMINAL_STATUSES:
            break
        time.sleep(2)

    print(f"Execution finished with status: {status}")

if __name__ == "__main__":
    main()

Interactive agent with user messages

from asteroid_odyssey import (
    Configuration,
    ApiClient,
    AgentsApi,
    ExecutionApi,
    AgentsAgentExecuteAgentRequest,
    AgentsExecutionUserMessagesAddTextBody,
)
import time
import os

TERMINAL_STATUSES = {"completed", "failed", "cancelled"}

def main():
    config = Configuration(
        api_key={"ApiKeyAuth": os.environ.get("ASTEROID_API_KEY")}
    )
    client = ApiClient(config)

    # Start execution
    agents_api = AgentsApi(client)
    request = AgentsAgentExecuteAgentRequest(
        inputs={"task": "Book an appointment"}
    )
    response = agents_api.agent_execute_post(
        agent_id="your-agent-id",
        agents_agent_execute_agent_request=request
    )
    execution_id = response.execution_id
    print(f"Execution started: {execution_id}")

    execution_api = ExecutionApi(client)

    while True:
        result = execution_api.execution_get(execution_id=execution_id)
        status = result.status.value
        print(f"Status: {status}")

        if status in TERMINAL_STATUSES:
            print(f"Execution completed with status: {status}")
            break

        if status == "paused_by_agent":
            # Agent is waiting for user input
            print("Agent is waiting for input...")
            user_input = input("Your response: ")

            body = AgentsExecutionUserMessagesAddTextBody(text=user_input)
            execution_api.execution_user_messages_add(
                execution_id=execution_id,
                agents_execution_user_messages_add_text_body=body
            )
            print("Message sent!")

        time.sleep(2)

if __name__ == "__main__":
    main()

Upload files to an execution

from asteroid_odyssey import (
    Configuration,
    ApiClient,
    FilesApi,
)
import os

def upload_files_example():
    config = Configuration(
        api_key={"ApiKeyAuth": os.environ.get("ASTEROID_API_KEY")}
    )
    client = ApiClient(config)
    files_api = FilesApi(client)

    execution_id = "your-execution-id"

    # Upload a file from disk
    with open("document.pdf", "rb") as f:
        file_content = f.read()

    files_api.execution_context_files_upload(
        execution_id=execution_id,
        files=[("document.pdf", file_content)]
    )
    print("File uploaded successfully!")

    # Upload multiple files
    files_to_upload = []
    for filename in ["file1.txt", "file2.txt"]:
        with open(filename, "rb") as f:
            files_to_upload.append((filename, f.read()))

    files_api.execution_context_files_upload(
        execution_id=execution_id,
        files=files_to_upload
    )
    print(f"Uploaded {len(files_to_upload)} files")

if __name__ == "__main__":
    upload_files_example()

Create and use an agent profile

from asteroid_odyssey import (
    Configuration,
    ApiClient,
    AgentsApi,
    AgentProfilesApi,
    AgentsAgentExecuteAgentRequest,
    AgentsProfileCreateAgentProfileRequest,
    AgentsProfileCountryCode,
    AgentsProfileProxyType,
    AgentsProfileCookie,
)
import os

def profile_example():
    config = Configuration(
        api_key={"ApiKeyAuth": os.environ.get("ASTEROID_API_KEY")}
    )
    client = ApiClient(config)

    profiles_api = AgentProfilesApi(client)

    # Create a profile with proxy and cookies
    request = AgentsProfileCreateAgentProfileRequest(
        name="E-commerce Profile",
        organization_id="your-org-id",
        country_code=AgentsProfileCountryCode.US,
        proxy_type=AgentsProfileProxyType.RESIDENTIAL,
        captcha_solver_active=True,
        cookies=[
            AgentsProfileCookie(
                name="session",
                key="session_id",
                value="abc123",
                domain="example.com",
                secure=True,
                http_only=True,
            )
        ]
    )
    profile = profiles_api.agent_profiles_create(
        agents_profile_create_agent_profile_request=request
    )
    print(f"Created profile: {profile.id}")

    # Execute an agent with the profile
    agents_api = AgentsApi(client)
    exec_request = AgentsAgentExecuteAgentRequest(
        inputs={"product_url": "https://example.com/product/123"},
        agent_profile_id=profile.id
    )
    response = agents_api.agent_execute_post(
        agent_id="your-agent-id",
        agents_agent_execute_agent_request=exec_request
    )
    print(f"Execution started: {response.execution_id}")

if __name__ == "__main__":
    profile_example()

Configuration Options

from asteroid_odyssey import Configuration, ApiClient

# Basic configuration
config = Configuration(
    api_key={"ApiKeyAuth": "your-api-key"}
)

# Custom host (for self-hosted or development)
config = Configuration(
    host="https://custom-host.example.com/agents/v2",
    api_key={"ApiKeyAuth": "your-api-key"}
)

# With debug logging
config = Configuration(
    api_key={"ApiKeyAuth": "your-api-key"},
    debug=True
)

client = ApiClient(config)

Execution Statuses

Status	Description
starting	Execution is initializing
running	Agent is actively working
paused	Execution is paused
paused_by_agent	Agent is waiting for user input
awaiting_confirmation	Waiting for confirmation
completed	Successfully finished
failed	Execution failed
cancelled	Execution was cancelled

Error Handling

from asteroid_odyssey import ApiException

try:
    response = agents_api.agent_execute_post(
        agent_id="invalid-id",
        agents_agent_execute_agent_request=request
    )
except ApiException as e:
    print(f"API Error: {e.status} - {e.reason}")
    print(f"Body: {e.body}")

Cookbook Examples

More practical examples including complex workflows

API Reference

Complete API documentation with all available endpoints