Skip to main content

Documentation Index

Fetch the complete documentation index at: https://snowglobe.so/docs/llms.txt

Use this file to discover all available pages before exploring further.

​
Snowglobe SDK Integration Guide

Still on snowglobe<=0.4.x? See the documentation for previous versions here.

​
Overview

Snowglobe is a simulation engine designed for testing and evaluating AI agents and chatbots through automated conversation generation and analysis. This guide demonstrates how to integrate Snowglobe into your continuous integration (CI) pipeline and programmatic workflows for comprehensive agent testing.

​
Table of Contents

​
Prerequisites

Before integrating Snowglobe, ensure you have:
  • Python 3.11+ installed
  • Snowglobe SDK package (pip install snowglobe-sdk)
  • Valid API credentials (API key and Organization ID) from here.
  • OpenAI API key (or other supported LLM provider credentials)
  • Access to a Snowglobe control plane instance

​
Required Dependencies

import asyncio
from snowglobe.sdk import SnowglobeClient
from snowglobe.sdk.types import AgentCreateSchema, SimulationCreateSchema

​
Authentication Setup

​
Environment Variables

Set up your credentials as environment variables or configuration constants: 
X_API_KEY = "eyJhbM..."  # Your Snowglobe API key
X_SNOWGLOBE_ORG_ID = "orZu..."  # Your organization ID
OPENAI_API_KEY = "sk-prnZS..."  # Your LLM provider API key

​
Client Configuration

Initialize the Snowglobe client with proper authentication headers: 
control_plane_url = "https://api.snowglobe.guardrailsai.com"  # Or your hosted instance

client = SnowglobeClient(
    api_key=X_API_KEY,
    organization_id=X_SNOWGLOBE_ORG_ID,
    base_url=control_plane_url,
    follow_redirects=True
)

​
Core Components

​
1. Agent Creation

Agents represent the AI systems you want to test. Each agent requires a name, icon, and connection info describing how Snowglobe should reach your LLM. 

agent_config: AgentCreateSchema = {
    "name": "customer_support_agent",
    "description": "Customer support agent for Amatto's pizza restaurant",
    "icon": "pizza",
    "connection_info": {
        "provider": "OpenAI",
        "endpoint": "",
        "model_name": "openai/gpt-4o",
        "api_key_ref": "OPENAI_API_KEY",
        "system_prompt": "You are a helpful expert customer support agent for Amatto's pizza.",
        "extra_body": [],
        "extra_headers": [],
    },
}
Note: Agents using a custom code integration via snowglobe-connect require a two-step setup:
  1. Create the agent with the API as shown above.
  2. Map the agent’s ID in your snowglobe-connect deployment’s agents.json file.
  3. Run snowglobe-connect start.

​
2. Simulation Configuration

Simulations define how conversations are generated and evaluated against your agent. 
simulation_config: SimulationCreateSchema = {
    "name": "continuous integration simulation",
    "role": "Customer support agent for Amatto's pizza restaurant",
    "user_description": "",
    "use_cases": "",
    "generation_status": "pending",
    "evaluation_status": "pending",
    "validation_status": "pending",
    "source_data": {
        "docs": {
            "misc": [],
            "knowledge_base": [],
            "historical_data": []
        },
        "evaluation_configuration": {
            "No Financial Advice": {
                "id": "e5af8dee-6d8d-4144-b754-204d24879ec9",
                "name": "No Financial Advice",
                "version": 1,
                "metadata": {},
            },
        },
        "generation_configuration": {
            "max_topics": 1,
            "max_personas": 4,
            "branching_factor": 25,
            "max_conversations": 500,
            "max_conversation_length": 4,
            "continue_conversations_from_adapted_messages": False,
            "data_gen_mode": "coverage_focused_v3",
            "intent": "",
            "persona_topic_generators": [
                {
                    "name": "app_description_system_prompt",
                    "settings": {
                        "max_personas": 4
                    }
                }
            ],
            "min_conversation_length": 1
        }
    },
    "is_template": False
}

​
CI Integration Workflow

​
Step 1: Create and Configure Agent

async def create_agent(client, agent_config):
    """Create a new agent for testing."""
    agent = await client.agents.create_agent(agent_config)
    print(f"Agent created successfully: {agent.id}")
    return agent.id

​
Step 2: Launch Simulation

async def launch_simulation(client, agent_id, simulation_config):
    """Create and launch a new simulation."""
    config = {
        **simulation_config,
        "application_id": agent_id,
        "app_id": agent_id,
    }
    simulation = await client.simulations.create_simulation(config)
    simulation_id = simulation.id

    # Auto-approve personas for CI automation
    await client.simulations.update_simulation_settings(
        simulation_id, {"autoApprovePersonas": True}
    )
    return simulation_id

​
Step 3: Monitor Simulation Progress

import asyncio

async def wait_for_completion(client, simulation_id, timeout_minutes=20):
    """Poll simulation until completion or timeout."""
    max_attempts = timeout_minutes * 6  # poll every 10 seconds
    poll_interval = 10

    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")
            return True

        await asyncio.sleep(poll_interval)

    raise TimeoutError("Simulation timed out before completion")

​
Step 4: Retrieve Results

import json

async def download_results(client, simulation_id):
    """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 filename

​
Complete CI Integration Example

import asyncio
import json
import os

from snowglobe.sdk import SnowglobeClient

async def run_agent_simulation(agent_config, simulation_config):
    """Complete workflow for running agent simulations in CI."""
    client = SnowglobeClient(
        api_key=os.environ["SNOWGLOBE_API_KEY"],
        organization_id=os.environ.get("SNOWGLOBE_ORG_ID"),
    )

    try:
        agent_id = await create_agent(client, agent_config)
        simulation_id = await launch_simulation(client, agent_id, simulation_config)
        await wait_for_completion(client, simulation_id)
        results_file = await download_results(client, simulation_id)

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

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


asyncio.run(
    run_agent_simulation(agent_config, simulation_config)
)

​
Error Handling

async def robust_simulation_run(agent_config, simulation_config):
    """Simulation run with comprehensive error handling."""
    import httpx

    try:
        return await run_agent_simulation(agent_config, simulation_config)
    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)}

​
CI Pipeline Integration

# Example GitHub Actions workflow
name: Agent Testing
on: [push, pull_request]

jobs:
  test-agent:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install dependencies
        run: pip install snowglobe-sdk
      - name: Run agent simulation
        env:
          SNOWGLOBE_API_KEY: ${{ secrets.SNOWGLOBE_API_KEY }}
          SNOWGLOBE_ORG_ID: ${{ secrets.SNOWGLOBE_ORG_ID }}
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: python test_agent_simulation.py

​
Simulation States

The state_num field on a simulation indicates its current phase:
state_numState nameDescription
0–2Draft / QueuedSimulation created, waiting to start
3–5Experiment startedInitialization and setup
6–8Generation in progressPersona, topic, and conversation generation
9–11Evaluation in progressAgent responses being judged against risks
12–14Validation in progressResults validated
15–16Adaptation in progressAdapted (adversarial) conversations being generated
17+Experiment completedResults available; download_simulation_data can be called
Poll sim.state_num and wait for >= 17 before downloading results.

​
Troubleshooting

​
Common Issues

Authentication errors
  • Verify your API key and organization ID are correct.
  • Confirm the x-api-key header is being sent (the SDK sets this automatically from api_key).
  • Check network connectivity to your control plane URL.
Simulation failures
  • Review the agent’s connection_info — all required fields (api_key_ref, model_name, system_prompt) must be present.
  • Verify the LLM provider API key referenced by api_key_ref is valid and has sufficient quota.
  • Check that source_data.generation_configuration values are within acceptable ranges.
Timeout issues
  • Increase timeout_minutes in wait_for_completion for large simulations.
  • Reduce max_personas, max_topics, or branching_factor for faster runs.

​
Debugging Tips

async def debug_simulation(client, simulation_id):
    """Print detailed simulation status for debugging."""
    sim = await client.simulations.get_simulation(simulation_id)

    print(f"State:             {sim.state} (num={sim.state_num})")
    print(f"Generation status: {sim.generation_status}")
    print(f"Evaluation status: {sim.evaluation_status}")
    print(f"Validation status: {sim.validation_status}")
    print(f"Status reason:     {sim.status_reason}")
    print(f"Statistics:        {sim.statistics}")