> ## 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.

# Integrate with CI/CD

> Automating Snowglobe testing in your development pipeline

# Snowglobe SDK Integration Guide

<Note>Still on `snowglobe<=0.4.x`?  See the documentation for previous versions [here](/snowglobe/docs/guide/integrate-with-cicd-v04x).</Note>

## 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](#prerequisites)
* [Authentication Setup](#authentication-setup)
* [Core Components](#core-components)
* [CI Integration Workflow](#ci-integration-workflow)
* [Troubleshooting](#troubleshooting)

## 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](https://snowglobe.so/app/keys).
* OpenAI API key (or other supported LLM provider credentials)
* Access to a Snowglobe control plane instance

### Required Dependencies

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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.

```python theme={null}

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` require a two-step setup:
>
> 1. Create the agent with the API as shown above.
> 2. Map the agent's ID in your `snowglobe` deployment's `agents.json` file.
> 3. Run `snowglobe connect --simulation-id <simulation-id>` for an existing simulation, or `snowglobe simulate ./config.yml` to create one and connect in a single step. (As of client `1.0.0`, `connect` replaces the previous `start` command and `--simulation-id` is required.)

### 2. Simulation Configuration

Simulations define how conversations are generated and evaluated against your agent.

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```yaml theme={null}
# 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_num` | State name             | Description                                                 |
| ----------- | ---------------------- | ----------------------------------------------------------- |
| 0–2         | Draft / Queued         | Simulation created, waiting to start                        |
| 3–5         | Experiment started     | Initialization and setup                                    |
| 6–8         | Generation in progress | Persona, topic, and conversation generation                 |
| 9–11        | Evaluation in progress | Agent responses being judged against risks                  |
| 12–14       | Validation in progress | Results validated                                           |
| 15–16       | Adaptation in progress | Adapted (adversarial) conversations being generated         |
| 17+         | Experiment completed   | Results 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

```python theme={null}
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}")
```