Metadata-Version: 2.5
Name: sketricgen
Version: 0.3.0
Summary: Python SDK for SketricGen Chat Server API
Project-URL: Homepage, https://sketric.com
Project-URL: Documentation, https://docs.sketric.com/sdk
Project-URL: Repository, https://github.com/sketric/sketric-sdk-python
Author-email: Sketric Solutions <dev@sketric.com>
License: MIT
Keywords: ai,api,chat,sdk,sketricgen,workflow
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: httpx>=0.25.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: typing-extensions>=4.0.0
Provides-Extra: dev
Requires-Dist: black>=23.0.0; extra == 'dev'
Requires-Dist: mypy>=1.0.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21.0; extra == 'dev'
Requires-Dist: pytest-mock>=3.10.0; extra == 'dev'
Requires-Dist: pytest>=7.0.0; extra == 'dev'
Requires-Dist: respx>=0.20.0; extra == 'dev'
Requires-Dist: ruff>=0.1.0; extra == 'dev'
Description-Content-Type: text/markdown

# SketricGen SDK

Python SDK for the SketricGen runtime and Admin APIs.

## Installation

```bash
pip install sketricgen
```

Or install from source:

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

## Quick Start

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

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

```python
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](https://docs.ag-ui.com) 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:

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

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

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

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

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

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

```python
# 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:

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

```python
# 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

```python
# 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:

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