Quickstart

pip install snowglobe-sdk
# or, if you use uv:
uv pip install snowglobe-sdk

import asyncio
import httpx
import json
import os

from snowglobe.sdk import SnowglobeClient

async def run_agent_simulation(agent_config, simulation_config):
    ### Step 1: Initialize the Snowglobe Client
    client = SnowglobeClient(
        api_key=os.environ["SNOWGLOBE_API_KEY"],
        organization_id=os.environ.get("SNOWGLOBE_ORG_ID"),
    )

    try:
        ### Step 2: Create Agent
        agent = await client.agents.create_agent({
            "name": "customer_support_agent",
            "description": "Customer support agent for Amatto's pizza restaurant",
            "icon": "pizza",
        })
        agent_id = agent.id
        

        ### Step 3: Create and Launch Simulation
        simulation = await client.simulations.create({
            name: "continuous integration simulation",
            agent_id: "00000000-0000-0000-0000-000000000000",
            max_conversation_length: 5,
            max_personas: 10,
            max_conversations: 50,
        })
        simulation_id = simulation.id
        
        
        ### Step 4: Poll simulation until completion or timeout.
        timeout_minutes = 20
        max_attempts = timeout_minutes * 6  # poll every 10 seconds
        completed = False

        for _ in range(max_attempts):
            sim = await client.simulations.get_simulation(simulation_id)
            state_num = sim.state_num
            print(f"Simulation state: {state_num}")

            # state_num >= 17 indicates completion (see Simulation States below)
            if state_num is not None and state_num >= 17:
                print("Simulation completed successfully")
                completed = True
                break

            await asyncio.sleep(10)

        if not completed:
            raise TimeoutError("Simulation timed out before completion")


        ### Step 5: Download simulation results for analysis.
        data = await client.simulations.download_simulation_data(simulation_id)

        filename = f"{simulation_id}_results.json"
        with open(filename, "w") as f:
            json.dump([d.model_dump() for d in data], f, indent=2)

        print(f"Results saved to {filename}")

        return {
            "success": True,
            "agent_id": agent_id,
            "simulation_id": simulation_id,
            "results_file": filename,
        }

    except Exception as e:
        print(f"Simulation failed: {e}")
        return {"success": False, "error": str(e)}


async def robust_simulation_run():
    """Simulation run with comprehensive error handling."""
    try:
        return await run_agent_simulation()
    except ValidationError as e:
        print(f"Configuration validation failed: {e}")
        return {"success": False, "error": "validation", "details": str(e)}
    except httpx.HTTPStatusError as e:
        print(f"API error {e.response.status_code}: {e.response.text}")
        return {"success": False, "error": "api_error", "details": str(e)}
    except TimeoutError as e:
        print(f"Simulation timed out: {e}")
        return {"success": False, "error": "timeout", "details": str(e)}
    except Exception as e:
        print(f"Unexpected error: {e}")
        return {"success": False, "error": "unexpected", "details": str(e)}

asyncio.run(
    robust_simulation_run()
)