Metadata-Version: 2.4
Name: snowglobe-sdk
Version: 1.0.1a2
Summary: SDK for the Snowglobe API
Author-email: Guardrails AI <contact@guardrailsai.com>
License: MIT License
        
        Copyright (c) 2024 Guardrails AI
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
        
Requires-Python: <4,>=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.27.0
Requires-Dist: pydantic<3.0.0,>=2.0.0
Requires-Dist: tenacity>=9.0.0
Provides-Extra: dev
Requires-Dist: ruff; extra == "dev"
Requires-Dist: pre-commit>=4.1.0; extra == "dev"
Requires-Dist: coverage>=7.6.12; extra == "dev"
Requires-Dist: pyright[nodejs]>=1.1.396; extra == "dev"
Dynamic: license-file

# Snowglobe Python SDK

Python SDK for the [Snowglobe](https://guardrailsai.com/snowglobe) API.

## Requirements

- Python 3.11+

## Installation

```bash
pip install snowglobe-sdk
```

##  Quickstart

For a condensed quickstart guide, see https://guardrailsai.com/snowglobe/docs/sdk/python/quickstart

## Initialization

```python
import asyncio
from snowglobe.sdk import SnowglobeClient

client = SnowglobeClient(
    api_key="your-api-key",
)
```

**With an organization ID** (multi-tenant):

```python
client = SnowglobeClient(
    api_key="your-api-key",
    organization_id="org-123",
)
```

**With a custom `httpx.AsyncClient`** (e.g. to configure timeouts or proxies):

```python
import httpx
from snowglobe.sdk import SnowglobeClient

http = httpx.AsyncClient(timeout=30.0)
client = SnowglobeClient(api_key="your-api-key", http_client=http)
```

All methods are `async` and must be awaited inside an async context.

```python
async def main():
    client = SnowglobeClient(api_key="your-api-key")
    agents = await client.agents.list_agents()
    print(agents)

asyncio.run(main())
```

---

## Namespaces

The client exposes four namespaced API objects:

| Namespace | Attribute | Description |
|---|---|---|
| Agents | `client.agents` | Create and manage AI agents (applications) |
| Simulations | `client.simulations` | Run and inspect simulations |
| Metrics | `client.metrics` | Browse risk/metric definitions |

---

## `client.agents`

Manage AI agents (applications under test).

### `create_agent(body)`

Create a new agent.

```python
agent = await client.agents.create_agent({
    "name": "My Chatbot",
    "icon": "🤖",
    "description": "Customer support bot",
})
print(agent.id)
```

### `list_agents()`

List all agents in the organization.

```python
agents = await client.agents.list_agents()
for agent in agents:
    print(agent.id, agent.name)
```

### `get_agent(id)`

Fetch a single agent by ID.

```python
agent = await client.agents.get_agent("agent-123")
```

### `update_agent(id, body)`

Update an existing agent.

```python
agent = await client.agents.update_agent("agent-123", {
    "name": "Updated Name",
    "description": "New description",
})
```

### `delete_agent(id)`

Delete an agent.

```python
await client.agents.delete_agent("agent-123")
```

---

## `client.simulations`

Run simulations and retrieve results.

### `create_simulation(body, *, as_draft=None)`

**DEPRECATED:** Use [`create(config)`](#createconfig) instead

Create a new simulation.

```python
simulation = await client.simulations.create_simulation({
    "name": "Q4 Red Team",
    "role": "assistant",
    "is_template": False,
    "risks": [{"id": "risk-123", "name": "Bias", "type": "LLM", "version": 1}],
})
print(simulation.id, simulation.state)
```

Pass `as_draft="true"` to save without launching.

### `create(config)`

Create and launch a simulation using a structured config object. This is the recommended way to start a new simulation.

```python
from snowglobe.sdk.models import SimulationCreateConfig

simulation = await client.simulations.create(SimulationCreateConfig(
    name="Q4 data privacy check",
    agent_id="agent-123",
    max_conversation_length=10,
    max_personas=20,
    max_conversations=5,
    simulation_prompt="Focus on data privacy edge cases.",
    agent_profile_id="profile-abc",  # optional
))
print(simulation.id, simulation.state)
```

You can also pass a plain dict — it will be validated against `SimulationCreateConfig`:

```python
simulation = await client.simulations.create({
    "name": "Q4 Probe",
    "agent_id": "agent-123",
    "max_conversation_length": 10,
    "max_personas": 20,
    "max_conversations": 5,
})
```

**`SimulationCreateConfig` fields:**

| Field | Type | Required | Description |
|---|---|---|---|
| `name` | `str` | Yes | Display name for the simulation |
| `agent_id` | `str` | Yes | ID of the agent to simulate against |
| `max_conversation_length` | `int` | Yes | Maximum number of turns per conversation |
| `max_personas` | `int` | Yes | Number of synthetic personas to generate |
| `max_conversations` | `int` | Yes | Number of conversations per persona |
| `agent_description` | `str` | No | Natural language description of the agent |
| `agent_profile_id` | `str` | No | ID of an agent profile to attach |
| `simulation_prompt` | `str` | No | Additional instructions for the simulation |

### `replay(config)`

Create and launch a simulation that replays existing conversations grouped by failure mode. Use this to verify that a previous class of failures has been resolved.

```python
from snowglobe.sdk.models import SimulationReplayConfig, TraceRegressionGroup

simulation = await client.simulations.replay(SimulationReplayConfig(
    name="Regression – data privacy fixes",
    agent_id="agent-123",
    conversations=[
        TraceRegressionGroup(
            conversation_ids=["conv-1", "conv-2", "conv-3"],
            failure_mode="data_privacy",
        ),
        TraceRegressionGroup(
            conversation_ids=["conv-4"],
            failure_mode="hallucination",
        ),
    ],
    agent_profile_id="profile-abc",  # optional
))
print(simulation.id)
```

You can also pass a plain dict:

```python
simulation = await client.simulations.replay({
    "name": "Regression – data privacy fixes",
    "agent_id": "agent-123",
    "conversations": [
        {"conversation_ids": ["conv-1", "conv-2"], "failure_mode": "data_privacy"},
    ],
})
```

**`SimulationReplayConfig` fields:**

| Field | Type | Required | Description |
|---|---|---|---|
| `name` | `str` | Yes | Display name for the simulation |
| `agent_id` | `str` | Yes | ID of the agent to simulate against |
| `conversations` | `list[TraceRegressionGroup]` | Yes | Groups of conversation IDs to replay |
| `agent_profile_id` | `str` | No | ID of an agent profile to attach |

**`TraceRegressionGroup` fields:**

| Field | Type | Description |
|---|---|---|
| `conversation_ids` | `list[str]` | IDs of existing conversations to replay |
| `failure_mode` | `str` | Label describing the failure class being retested |

### `list_simulations(*, limit=None)`

List simulations, optionally capped at `limit` results.

```python
simulations = await client.simulations.list_simulations(limit=20)
```

### `get_simulation(id, *, exclude_calculated_status=None)`

Fetch a simulation by ID.

```python
sim = await client.simulations.get_simulation("sim-123")
print(sim.state, sim.statistics)
```

### `update_simulation(id, body, *, as_draft=None)`

Update a simulation's configuration.

```python
sim = await client.simulations.update_simulation("sim-123", {
    "name": "Updated Sim Name",
})
```

### `delete_simulation(id)`

Delete a simulation.

```python
await client.simulations.delete_simulation("sim-123")
```

### `update_simulation_settings(id, body, *, access=None)`

Update settings (e.g. auto-approve personas) for a simulation.

```python
settings = await client.simulations.update_simulation_settings("sim-123", {
    "autoApprovePersonas": True,
})
```

### `download_simulation_data(id)`

Download the full conversation dataset for a completed simulation.

```python
data = await client.simulations.download_simulation_data("sim-123")
for conversation in data:
    print(conversation.persona, len(conversation.messages), "turns")
```

### `get_test(id, test_id, *, include_embedding=None)`

Fetch a single test (conversation turn) with its risk evaluations.

```python
test = await client.simulations.get_test("sim-123", "test-456")
print(test.prompt, test.response)
for evaluation in test.risk_evaluations:
    print(evaluation.risk_type, evaluation.risk_triggered)
```

### `batch_create_risk_evaluations(id, test_id, body, *, tests=None)`

Submit risk evaluation results for a test in bulk.

```python
evaluations = await client.simulations.batch_create_risk_evaluations(
    "sim-123",
    "test-456",
    [
        {
            "risk_type": "bias",
            "risk_triggered": False,
            "confidence": 95,
            "judge_response": "No bias detected.",
        }
    ],
)
```

### `list_conversations_for_test(id, test_id, *, start_from=None, app_id=None, include_adaptability_messages=None)`

List the full conversation history associated with a test.

```python
conversations = await client.simulations.list_conversations_for_test(
    "sim-123", "test-456"
)
for conv in conversations:
    for msg in conv.messages:
        print(f"[{msg.role}] {msg.content}")
```

### `get_simulation_health(id)`

Fetch the health details for a simulation.

```python
simulation_health = await client.simulations.get_simulation_health("sim-123")

print(f"Health Status: {simulation_health.health.status}")

print(f"Conversation Breakdown:")

print(f"  ==> Total Conversations: {simulation_health.health.generation.progress.conversations_total}")

print(f"  ==> Successfully Completed Conversations: {simulation_health.health.generation.progress.conversations_complete}")

print(f"  ==> Failed Conversations: {simulation_health.health.generation.progress.conversations_failed}")

print(f"  ==> Conversation Issues:")

for idx, iss in enumerate(simulation_health.health.issues):
    print(f"    {idx}. {iss.affected_count} {iss.title} ({iss.affected_pct}% of total conversations)")
    print(f"        - {iss.recommended_action}")

# The above would produce output like the below example
"""
Health Status: degraded
Conversation Breakdown:
  ==> Total Conversations: 5
  ==> Successfully Completed Conversations: 3
  ==> Failed Conversations: 2
  ==> Conversation Issues:
    1. 1 Schema validation errors (20.0% of total conversations)
        - Review the affected conversations and check whether the target response schema or tool mock output changed.
    2. 1 Service limit errors (20.0% of total conversations)
        - Review the agent implementation and prompts to ensure it stays within the service limit quotas.
"""

```

---

## `client.metrics`

Browse risk/metric definitions available in the platform.

### `list_metrics(*, lineage_id=None, version=None)`

List all available risk metrics, optionally filtered.

```python
risks = await client.metrics.list_metrics()
for risk in risks:
    print(risk.name, risk.type, f"v{risk.version}")
```

Filter by lineage:

```python
risks = await client.metrics.list_metrics(lineage_id="lineage-abc", version="2")
```

### `get_metric(id)`

Fetch a single risk metric by ID.

```python
risk = await client.metrics.get_metric("risk-123")
print(risk.name, risk.promptSource)
```
---

## Return Types

All methods return instances of the method's respective return type as a pydantic model. The full model definitions live in `snowglobe.sdk.models` and can be imported directly for type annotations.

```python
from snowglobe.sdk.models import Agent, Simulation, Risk
```

## Error Handling

Network or API errors raise `httpx.HTTPStatusError`. All methods automatically retry up to 5 times with exponential backoff (powered by [tenacity](https://tenacity.readthedocs.io)).

```python
import httpx

try:
    agent = await client.agents.get_agent("nonexistent-id")
except httpx.HTTPStatusError as e:
    print(e.response.status_code, e.response.text)
```

---

## Simulation States

The `state_num` field on a simulation indicates its current phase.  The state will give you a high level indication of the simulations progress:

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

---

## Simulation Health

The Simulation Health API provides detailed insights into the progress of a simulation.  Health is tracked progressively over time and thus can be called throughout and after the lifecycle of a simulation.  You can retrieve the simulation health by calling `client.simulations.get_simulation_health(simulation_id)`.

The resposne has two top level properties:
1. `health` a detailed and aggregated view of the simulations health and any issues that occurred during its lifecycle.
2. `health_details` the raw health stats recorded during the simulation's lifecycle. This includes a sample, but not all, of the issues encountered.

See [the get_simulation_health section](#get_simulation_healthid) for an example of how to use this data.
---

## 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**
- 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
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}")
```
