snowglobe-sdk 1.0.1a0

SDK for the Snowglobe API

Snowglobe Python SDK

Python SDK for the Snowglobe API.

Requirements

• Python 3.11+

Installation

pip install snowglobe-sdk

Quickstart

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

Initialization

import asyncio
from snowglobe.sdk import SnowglobeClient

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

With an organization ID (multi-tenant):

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

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

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.

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.

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.

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

get_agent(id)

Fetch a single agent by ID.

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

update_agent(id, body)

Update an existing agent.

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

delete_agent(id)

Delete an agent.

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

---

client.simulations

Run simulations and retrieve results.

create_simulation(body, *, as_draft=None)

DEPRECATED: Use create(config) instead

Create a new simulation.

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.

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:

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.

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:

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.

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

get_simulation(id, *, exclude_calculated_status=None)

Fetch a simulation by ID.

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.

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

delete_simulation(id)

Delete a simulation.

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

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

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

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

download_simulation_data(id)

Download the full conversation dataset for a completed simulation.

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.

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.

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.

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.

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.

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

Filter by lineage:

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

get_metric(id)

Fetch a single risk metric by ID.

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.

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

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

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}")