SketricGen SDK

Python SDK for the SketricGen runtime and Admin APIs.

Installation

pip install sketricgen

Or install from source:

git clone https://github.com/sketricsolutions/sketricgen-sdk.git
cd sketricgen-sdk
pip install -e .

Quick Start

from sketricgen import SketricGenClient

# Initialize client
client = SketricGenClient(api_key="sk_api_...")

# Run a workflow
response = await client.run_workflow(
    agent_id="agent-123",
    user_input="Hello, how are you?",
)
print(response.response)

Features

• Run Workflow: Execute chat/workflow requests with agents
• Control Plane: Manage projects, agents, knowledge bases, brand agents, and connectors through the same client
• Human in the Loop: Enable, inspect, and resume HITL requests
• Streaming: Real-time streaming responses using Server-Sent Events
• File Attachments: Attach images, PDFs, spreadsheets, text, structured data, and source files
• Async & Sync: Both async and synchronous API support
• Type Safety: Full type hints for IDE support
• Error Handling: Comprehensive custom exception types

Usage Examples

Non-Streaming Workflow

from sketricgen import SketricGenClient

client = SketricGenClient(api_key="your-api-key")

# Async
response = await client.run_workflow(
    agent_id="agent-123",
    user_input="What is the weather like today?",
    conversation_id="conv-456",  # Optional: resume conversation
)
print(f"Response: {response.response}")
print(f"Conversation ID: {response.conversation_id}")

# Sync
response = client.run_workflow_sync(
    agent_id="agent-123",
    user_input="Hello!",
)

Streaming Workflow

import json
from sketricgen import SketricGenClient

client = SketricGenClient(api_key="your-api-key")

# Async streaming
async for event in await client.run_workflow(
    agent_id="agent-123",
    user_input="Tell me a story",
    stream=True,
):
    data = json.loads(event.data)
    event_type = data["type"]
    
    if event_type == "TEXT_MESSAGE_CONTENT":
        # Print text chunks as they arrive
        print(data["delta"], end="", flush=True)
    elif event_type == "TOOL_CALL_START":
        print(f"\n[Calling tool: {data['tool_call_name']}]")
    elif event_type == "TOOL_CALL_END":
        print(f"[Tool completed]")
    elif event_type == "RUN_FINISHED":
        print()  # New line
    elif event_type == "RUN_ERROR":
        print(f"\nError: {data['message']}")

# Sync streaming
for event in client.run_workflow_sync(
    agent_id="agent-123",
    user_input="Tell me a story",
    stream=True,
):
    data = json.loads(event.data)
    if data["type"] == "TEXT_MESSAGE_CONTENT":
        print(data["delta"], end="", flush=True)

Stream Event Types (AG-UI Protocol):

The streaming API uses AG-UI events from ag_ui.core:

Event Type	Description	Key Fields
RUN_STARTED	Workflow execution started	thread_id, run_id
TEXT_MESSAGE_START	Assistant message started	message_id, role
TEXT_MESSAGE_CONTENT	Text chunk	message_id, delta
TEXT_MESSAGE_END	Assistant message completed	message_id
TOOL_CALL_START	Tool/function call started	tool_call_id, tool_call_name
TOOL_CALL_END	Tool/function call completed	tool_call_id
RUN_FINISHED	Workflow completed	thread_id, run_id, result
RUN_ERROR	Workflow error occurred	message
RUN_PAUSED_HITL	Workflow paused for a human decision	request_id, interrupt_ids
CUSTOM_EVENT	Custom event, including deep_agents_hitl	varies

RUN_FINISHED, RUN_ERROR, and RUN_PAUSED_HITL are terminal events; check event.is_terminal when consuming a stream.

Human-in-the-Loop

HITL is opt-in for API callers. Send enable_hitl=True on the initial turn and on every resume turn:

from sketricgen import HitlDecision, HitlResume, SketricGenClient

client = SketricGenClient(api_key="sk_api_...")
paused = await client.run_workflow(
    agent_id="agent-123",
    user_input="Schedule a recurring report",
    enable_hitl=True,
)

if paused.run_paused_hitl and paused.hitl_request:
    response = await client.run_workflow(
        agent_id="agent-123",
        conversation_id=paused.conversation_id,
        enable_hitl=True,
        hitl_resume=HitlResume(
            request_id=paused.hitl_request.request_id,
            decisions=[HitlDecision(type="approve")],
        ),
    )

Decisions are positional: provide one respond, approve, or reject decision for each action, in the order returned by hitl_request.action_requests.

Workflow with File Attachments

Attach files to your workflows. The SDK handles file uploads automatically in the background.

from sketricgen import SketricGenClient

client = SketricGenClient(api_key="your-api-key")

# Async with file attachment
response = await client.run_workflow(
    agent_id="agent-123",
    user_input="Please analyze this document",
    file_paths=["/path/to/document.pdf"],
)
print(response.response)

# Sync with file attachment
response = client.run_workflow_sync(
    agent_id="agent-123",
    user_input="Summarize this document",
    file_paths=["/path/to/document.pdf"],
)

Multiple File Attachments

from sketricgen import SketricGenClient

client = SketricGenClient(api_key="your-api-key")

# Attach multiple files at once
response = await client.run_workflow(
    agent_id="agent-123",
    user_input="Compare these two documents",
    file_paths=[
        "/path/to/document1.pdf",
        "/path/to/document2.pdf",
    ],
)
print(response.response)

Error Handling

from sketricgen import (
    SketricGenClient,
    SketricGenAPIError,
    SketricGenAuthenticationError,
    SketricGenValidationError,
    SketricGenNetworkError,
    SketricGenFileSizeError,
    SketricGenContentTypeError,
)

client = SketricGenClient(api_key="your-api-key")

try:
    response = await client.run_workflow(
        agent_id="agent-123",
        user_input="Analyze this document",
        file_paths=["/path/to/file.pdf"],
    )
except SketricGenAuthenticationError as e:
    print(f"Authentication failed: {e}")
except SketricGenFileSizeError as e:
    print(f"File too large: {e}")
    print(f"Max size: {e.max_size} bytes")
except SketricGenContentTypeError as e:
    print(f"Unsupported file type: {e}")
    print(f"Allowed types: {e.allowed_types}")
except SketricGenValidationError as e:
    print(f"Validation error: {e}")
except SketricGenAPIError as e:
    print(f"API error ({e.status_code}): {e}")
except SketricGenNetworkError as e:
    print(f"Network error: {e}")
except FileNotFoundError as e:
    print(f"File not found: {e}")

Configuration

from sketricgen import SketricGenClient

# Direct configuration
client = SketricGenClient(
    api_key="your-api-key",
    timeout=30,
    upload_timeout=300,  # 5 minutes for large files
    max_retries=3,
)

# From environment variables
# Set SKETRICGEN_API_KEY
client = SketricGenClient.from_env()

Supported File Types

File attachments support JPEG, PNG, WebP, GIF, PDF, CSV, XLS/XLSX, JSON, HTML, Markdown, plain text, XML, YAML, and common source-code formats including JavaScript, TypeScript, CSS, Python, Ruby, PHP, Java, Kotlin, Go, Rust, Swift, Scala, C/C++, C#, SQL, and shell scripts.

Maximum file size: 20 MB

API Reference

SketricGenClient

run_workflow(agent_id, user_input?, conversation_id?, contact_id?, file_paths?, stream?, enable_hitl?, hitl_resume?)

Execute a workflow/chat request.

Parameters:

• agent_id (str): Agent ID to chat with
• user_input (str, optional): User message (max 10000 characters); omit only for HITL resume or asset-only runs
• conversation_id (str, optional): Conversation ID for resuming
• contact_id (str, optional): External contact ID
• file_paths (list[str], optional): List of file paths to upload and attach
• stream (bool, optional): Whether to stream the response
• enable_hitl (bool, optional): Enable HITL tools for this turn
• hitl_resume (HitlResume, optional): Decisions for a pending HITL request

Returns: ChatResponse or AsyncIterator[StreamEvent] if streaming

Response Models

ChatResponse

• agent_id: Workflow ID
• user_id: User identifier
• conversation_id: Conversation ID
• response: Assistant's response
• owner: Owner of the agent
• error: Error flag
• run_paused_hitl: Whether the run paused for human input
• hitl_request: Typed pending request metadata required to resume

StreamEvent

• event_type: SSE envelope event type; AG-UI type is in the JSON data
• data: Event content
• id: Optional event ID
• is_terminal: True for finished, failed, or HITL-paused runs

Sync Methods

The async run_workflow() method has a synchronous variant:

• run_workflow_sync()

Control Plane

SketricGenClient also manages teamspace resources through the Admin API. A single sk_api_… key may carry runtime, admin, or both accesses. Calls fail with the API's authorization error when the key lacks the required access.

Construction

from sketricgen import SketricGenClient

# Reads SKETRICGEN_API_KEY
client = SketricGenClient.from_env()

Every method is async-first with a synchronous _sync twin, matching the runtime methods.

whoami

# Async
me = await client.whoami()
print(me.teamspace_id, me.display_name)

# Sync
me = client.whoami_sync()

Listing resources

projects.list(), agents.list(), and knowledge_bases.list() transparently page through every result — you never manage a next_token cursor. They return a lazy iterator, so only the pages you consume are fetched:

# Async — lazy async iterator
async for project in client.projects.list():
    print(project.project_id, project.display_name)

async for agent in client.agents.list():
    print(agent.agent_id, agent.name)

async for kb in client.knowledge_bases.list():
    print(kb.knowledge_base_id, kb.name)

# Sync — lazy iterator
for project in client.projects.list_sync():
    print(project.project_id)
for agent in client.agents.list_sync():
    print(agent.agent_id)
for kb in client.knowledge_bases.list_sync():
    print(kb.knowledge_base_id)

Brand agents

Brand-agent provisioning is asynchronous on the server. Use create_and_wait() for the one-call path, or drive the create() / get_status() primitives yourself.

# One call: create and poll to completion
job = await client.brand_agents.create_and_wait(
    name="Acme Support",
    seed_url="https://acme.example.com",
    poll_interval=10.0,   # seconds between polls
    timeout=600.0,        # give up after this many seconds
)
print(job.status)              # "succeeded"
print(job.embed.snippet)       # embed code for the finished agent

# Or drive the async flow yourself
job = await client.brand_agents.create(
    name="Acme Support",
    seed_url="https://acme.example.com",
)
status = await client.brand_agents.get_status(job.job_id)
print(status.status, status.phase)

# Browse the template catalog
templates = await client.brand_agents.list_templates()

# Inspect and edit an existing brand agent
detail = await client.brand_agents.get(agent_id)
result = await client.brand_agents.update(
    agent_id,
    display_name="Acme Assistant",
    instructions="Be concise.",
    model="gpt-4o",
    knowledge_base_ids=["kb-123"],
)

# Widget configuration
config = await client.brand_agents.get_widget_config(agent_id)
await client.brand_agents.update_widget_config(agent_id, primary_color="#0055FF")

# Sync
job = client.brand_agents.create_and_wait_sync(
    name="Acme Support", seed_url="https://acme.example.com",
)
templates = client.brand_agents.list_templates_sync()

create_and_wait() raises SketricGenJobError if the job fails and SketricGenTimeoutError if it does not reach a terminal state within timeout — a failed provision is never silently treated as success. Sync twins: create_and_wait_sync(), create_sync(), get_status_sync(), get_sync(), update_sync(), list_templates_sync(), get_widget_config_sync(), update_widget_config_sync().

Connectors

# Curated brand-connector catalog
connectors = await client.connectors.list()

# Grantable tool names for one connector
tools = await client.connectors.list_tools("gmail")

# Mint a hosted consent URL, then poll until a human finishes connecting
link = await client.connectors.create_link("gmail", project_id="proj-123")
print(link.connect_url)
status = await client.connectors.check_connection("gmail", project_id="proj-123")
print(status.connected)

# Attach a connector's tools to an agent, or detach it
await client.connectors.attach(agent_id, "gmail", allowed_tools=["send_email"])
await client.connectors.detach(agent_id, "gmail")

# Sync
connectors = client.connectors.list_sync()
client.connectors.attach_sync(agent_id, "gmail", allowed_tools=["send_email"])

Attach/detach are grouped under connectors (connector-centric) even though the underlying route lives under /agents. Sync twins: list_sync(), list_tools_sync(), create_link_sync(), check_connection_sync(), attach_sync(), detach_sync().

Error handling

Control-plane errors carry a machine-readable code, so you can branch on the failure kind instead of parsing message strings:

from sketricgen import (
    SketricGenClient,
    SketricGenAdminError,
    SketricGenAuthenticationError,
    SketricGenJobError,
    SketricGenError,
)

client = SketricGenClient.from_env()

try:
    job = await client.brand_agents.create_and_wait(
        name="Acme", seed_url="https://acme.example.com",
    )
except SketricGenAuthenticationError:
    print("API key rejected (401)")
except SketricGenJobError as e:
    print(f"Provisioning failed for job {e.job_id}: {e.error_summary}")
except SketricGenAdminError as e:
    if e.code == "agent_limit_reached":
        print("Out of agent quota")
    else:
        print(f"Admin API error [{e.status_code}] {e.code}: {e}")
except SketricGenError as e:
    # Base class — catches anything the SDK raises, data plane or control plane
    print(f"SDK error: {e}")

Response models tolerate unknown fields, so a server-side field addition never breaks a pinned SDK version.

License

MIT License