Skip to main content

​
SketricGen Python SDK

Python SDK for the SketricGen Chat Server API. Build AI-powered workflows, chat applications, and document analysis tools.

​
Requirements

  • Python 3.9+
  • Dependencies: httpx>=0.25.0, pydantic>=2.0.0, typing-extensions>=4.0.0

​
Installation

pip install sketricgen
Or with other package managers: 
# uv
uv add sketricgen

# poetry
poetry add sketricgen

​
Quick Start

from sketricgen import SketricGenClient

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

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

​
Features

  • Run Workflows: Execute chat/workflow requests with AI agents
  • Streaming: Real-time streaming responses using Server-Sent Events
  • File Attachments: Attach images and PDFs to workflows
  • Async & Sync: Both async and synchronous API support
  • Type Safety: Full type hints for IDE support
  • Error Handling: Comprehensive custom exception types

​
Client Reference

​
SketricGenClient

The main client for interacting with the SketricGen API.

​
Constructor

client = SketricGenClient(
    api_key: str,           # Your SketricGen API key (required)
    timeout: int = 300,     # Request timeout in seconds
    upload_timeout: int = 300,  # Upload timeout for large files
    max_retries: int = 3,   # Maximum retry attempts
)

​
From Environment Variables

# Set SKETRICGEN_API_KEY environment variable
client = SketricGenClient.from_env()
Environment VariableDefaultDescription
SKETRICGEN_API_KEYrequiredYour API key
SKETRICGEN_TIMEOUT300Request timeout in seconds
SKETRICGEN_UPLOAD_TIMEOUT300Upload timeout for large files
SKETRICGEN_MAX_RETRIES3Maximum retry attempts

​
run_workflow()

Execute a workflow/chat request (async). 
response = await client.run_workflow(
    agent_id: str,                    # Agent ID to chat with (required)
    user_input: str,                  # User message, max 10000 chars (required)
    conversation_id: str = None,      # Resume a conversation
    contact_id: str = None,           # External contact ID
    file_paths: list[str] = None,     # Files to attach
    stream: bool = False,             # Enable streaming
)
Returns: ChatResponse if stream=False, AsyncIterator[StreamEvent] if stream=True

​
Example: Non-Streaming

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

​
Example: Streaming

import json

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)
    
    if data["type"] == "TEXT_MESSAGE_CONTENT":
        print(data["delta"], end="", flush=True)
    elif data["type"] == "RUN_FINISHED":
        print()

​
run_workflow_sync()

Synchronous version of run_workflow(). 
response = client.run_workflow_sync(
    agent_id: str,
    user_input: str,
    conversation_id: str = None,
    contact_id: str = None,
    file_paths: list[str] = None,
    stream: bool = False,
)
Returns: ChatResponse if stream=False, Iterator[StreamEvent] if stream=True

​
Example

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

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

​
Response Models

​
ChatResponse

Response from a non-streaming workflow request.
FieldTypeDescription
agent_idstrWorkflow/Agent ID
user_idstrUser identifier
conversation_idstrConversation ID for follow-up messages
responsestrThe assistant’s response text
ownerstrOwner of the agent
errorboolError flag (default: False)

​
StreamEvent

Individual event from a streaming response.
FieldTypeDescription
event_typestrType of the SSE event
datastrJSON string containing event data
idstr | NoneOptional event ID

​
Streaming Events

The streaming API uses the AG-UI Protocol. Parse the data field as JSON to access event details.

​
Event Types

Event TypeDescriptionKey Fields
RUN_STARTEDWorkflow execution startedthread_id, run_id
TEXT_MESSAGE_STARTAssistant message startedmessage_id, role
TEXT_MESSAGE_CONTENTText chunk receivedmessage_id, delta
TEXT_MESSAGE_ENDAssistant message completedmessage_id
TOOL_CALL_STARTTool/function call startedtool_call_id, tool_call_name
TOOL_CALL_ENDTool/function call completedtool_call_id
RUN_FINISHEDWorkflow completedthread_id, run_id
RUN_ERRORWorkflow error occurredmessage
CUSTOMCustom eventvaries

​
Complete Streaming Example

import json
from sketricgen import SketricGenClient

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

async for event in await client.run_workflow(
    agent_id="agent-123",
    user_input="Search for info and summarize it",
    stream=True,
):
    data = json.loads(event.data)
    event_type = data["type"]
    
    if event_type == "RUN_STARTED":
        print(f"Started run: {data.get('run_id')}")
    elif event_type == "TEXT_MESSAGE_CONTENT":
        print(data["delta"], end="", flush=True)
    elif event_type == "TOOL_CALL_START":
        print(f"\n[Calling: {data['tool_call_name']}]")
    elif event_type == "TOOL_CALL_END":
        print("[Done]")
    elif event_type == "RUN_FINISHED":
        print("\n\nCompleted!")
    elif event_type == "RUN_ERROR":
        print(f"\nError: {data['message']}")

​
File Attachments

Attach files to workflows for document analysis and image understanding.

​
Supported File Types

TypeMIME TypesMax Size
Imagesimage/jpeg, image/png, image/webp, image/gif20 MB
Documentsapplication/pdf20 MB

​
Single File

response = await client.run_workflow(
    agent_id="agent-123",
    user_input="Summarize this document",
    file_paths=["/path/to/document.pdf"],
)
print(response.response)

​
Multiple Files

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)

​
Streaming with Files

async for event in await client.run_workflow(
    agent_id="agent-123",
    user_input="Analyze this image",
    file_paths=["/path/to/image.png"],
    stream=True,
):
    data = json.loads(event.data)
    if data["type"] == "TEXT_MESSAGE_CONTENT":
        print(data["delta"], end="", flush=True)