sinas 0.1.3

Python SDK for SINAS - AI Agent Platform

SINAS Python SDK

Python SDK for SINAS - AI Agent & Automation Orchestration Platform.

Runtime API Client for executing agents, managing state, running webhooks, and tracking executions.

Installation

pip install sinas

With FastAPI integration:

pip install sinas[fastapi]

Quick Start

Authentication

from sinas import SinasClient

# Initialize client with base URL
client = SinasClient(base_url="http://localhost:51245")

# Step 1: Login with email (sends OTP)
login_response = client.auth.login("[e-mail]")
session_id = login_response["session_id"]

# Step 2: Verify OTP and get access + refresh tokens
response = client.auth.verify_otp(session_id, "123456")
# Tokens are automatically set on the client
access_token = response["access_token"]
refresh_token = response["refresh_token"]

# Or use an API key directly
client = SinasClient(
    base_url="http://localhost:51245",
    api_key="your-api-key"
)

Environment Variables

Configure the client using environment variables:

export SINAS_BASE_URL="http://localhost:51245"
export SINAS_API_KEY="your-api-key"
# or
export SINAS_TOKEN="your-jwt-token"

from sinas import SinasClient

# Automatically uses environment variables
client = SinasClient()

Core Features

State Management

Store and retrieve key-value data with namespacing and visibility controls:

# Create/set a state value
state = client.state.set(
    namespace="user_prefs",
    key="theme",
    value={"mode": "dark", "accent": "blue"},
    description="User theme preferences",
    visibility="private"  # private, group, or public
)

# Get a specific state
state = client.state.get(state["id"])
print(state["value"])  # {"mode": "dark", "accent": "blue"}

# List states with filtering
states = client.state.list(
    namespace="user_prefs",
    visibility="private",
    tags="settings",
    search="theme"
)

# Update state
updated = client.state.update(
    state["id"],
    value={"mode": "light", "accent": "green"}
)

# Delete state
client.state.delete(state["id"])

Agent Chats

Create chats with agents and send messages:

# Create a chat with an agent
chat = client.chats.create(
    namespace="customer-support",
    agent_name="cs-agent-v1",
    title="Customer Inquiry",
    input={"customer_id": "12345"}  # Optional input validated by agent
)

# Send a message (blocking)
response = client.chats.send(
    chat_id=chat["id"],
    content="What's the status of my order?"
)
print(response["content"])

# Stream a message (Server-Sent Events)
for chunk in client.chats.stream(
    chat_id=chat["id"],
    content="Tell me about your services"
):
    # Parse SSE data
    import json
    try:
        data = json.loads(chunk)
        if "content" in data:
            print(data["content"], end="", flush=True)
    except:
        pass

# Get chat with all messages
chat_data = client.chats.get(chat["id"])
messages = chat_data["messages"]

# List all chats
chats = client.chats.list()

# Update chat
client.chats.update(chat["id"], title="Updated Title")

# Delete chat
client.chats.delete(chat["id"])

Webhook Execution

Execute webhooks by path:

# Execute a webhook
result = client.webhooks.run(
    path="process-payment",
    method="POST",
    body={"amount": 100, "currency": "USD"},
    headers={"X-Custom-Header": "value"},
    query={"idempotency_key": "abc123"}
)

print(result["execution_id"])  # Track execution
print(result["result"])  # Function output

Executions

Track and manage function executions:

# List recent executions
executions = client.executions.list(
    function_name="payment-processor",
    status="completed",
    limit=10
)

# Get execution details
execution = client.executions.get("execution-id")
print(execution["status"])  # running, completed, failed, awaiting_input
print(execution["output_data"])

# Get execution steps
steps = client.executions.get_steps("execution-id")
for step in steps:
    print(f"{step['step_name']}: {step['status']}")

# Continue a paused execution
result = client.executions.continue_execution(
    "execution-id",
    input={"user_choice": "approve"}
)

Authentication Methods

# Refresh access token
refreshed = client.auth.refresh(refresh_token)
new_access_token = refreshed["access_token"]

# Exchange external OIDC token
response = client.auth.external_auth(external_oidc_token)

# Get current user info
user = client.auth.get_me()
print(user["email"])

# Check user permissions with OR logic (user needs at least one)
check = client.auth.check_permissions(
    ["sinas.functions.read:all", "sinas.functions.create:own"],
    logic="OR"
)
print(check["result"])  # True if user has at least one permission
print(check["checks"])  # Detailed results for each permission

# Check user permissions with AND logic (user needs all)
check = client.auth.check_permissions(
    ["sinas.admin:all", "sinas.functions.write:all"],
    logic="AND"
)
print(check["result"])  # True only if user has all permissions

# Logout (revoke refresh token)
client.auth.logout(refresh_token)

FastAPI Integration

SINAS provides two powerful approaches for FastAPI integration:

Approach 1: Ready-to-Mount Routers

Mount SINAS Runtime API endpoints directly in your FastAPI app with automatic authentication:

from fastapi import FastAPI
from sinas.integrations.routers import create_runtime_router

app = FastAPI()

# Mount ALL runtime endpoints at once
app.include_router(
    create_runtime_router("http://localhost:51245", include_auth=False),
    prefix="/api/runtime"
)

# This automatically creates endpoints like:
# - GET    /api/runtime/states
# - POST   /api/runtime/states
# - GET    /api/runtime/chats
# - POST   /api/runtime/agents/{namespace}/{agent}/chats
# - POST   /api/runtime/webhooks/{path}
# - GET    /api/runtime/executions

Or mount individual routers:

from sinas.integrations.routers import (
    create_state_router,
    create_chat_router,
    create_webhook_router,
    create_executions_router
)

app.include_router(create_state_router("http://localhost:51245"), prefix="/runtime/states")
app.include_router(create_chat_router("http://localhost:51245"), prefix="/runtime/chats")
app.include_router(create_webhook_router("http://localhost:51245"), prefix="/runtime/webhooks")
app.include_router(create_executions_router("http://localhost:51245"), prefix="/runtime/executions")

Approach 2: Custom Endpoints with SDK Client

Build custom endpoints with business logic using the authenticated client:

import asyncio
from fastapi import FastAPI, Depends
from sinas import SinasClient
from sinas.integrations.fastapi import SinasAuth

app = FastAPI()
sinas = SinasAuth(base_url="http://localhost:51245")

# Include auto-generated auth endpoints
app.include_router(sinas.router, prefix="/auth")

# Custom endpoint with auto-authentication
@app.get("/my-states")
async def get_my_states(client: SinasClient = Depends(sinas)):
    """List user's states with custom filtering."""
    states = await asyncio.to_thread(client.state.list, limit=10)
    return [{"id": s["id"], "key": s["key"]} for s in states]

# Combine multiple SDK calls
@app.post("/quick-chat")
async def quick_chat(
    agent_namespace: str,
    agent_name: str,
    message: str,
    client: SinasClient = Depends(sinas)
):
    """Create chat and send message in one request."""
    chat = await asyncio.to_thread(client.chats.create, agent_namespace, agent_name)
    response = await asyncio.to_thread(client.chats.send, chat["id"], message)
    return {"chat_id": chat["id"], "response": response}

# Permission-protected endpoint
@app.post("/admin/cleanup")
async def cleanup_states(
    namespace: str,
    client: SinasClient = Depends(sinas.require("sinas.contexts.delete:all"))
):
    """Admin-only: Clean up states (requires permission)."""
    states = await asyncio.to_thread(client.state.list, namespace=namespace)
    for state in states:
        await asyncio.to_thread(client.state.delete, state["id"])
    return {"deleted": len(states)}

Auto-Generated Auth Endpoints

The sinas.router automatically provides:

• POST /auth/login - Send OTP to email
• POST /auth/verify-otp - Verify OTP and get tokens
• POST /auth/refresh - Refresh access token using refresh token
• GET /auth/me - Get current authenticated user

Permission-Based Access Control

Use sinas.require() to protect endpoints with flexible AND/OR logic:

# Require a single permission
@app.delete("/states/{state_id}")
async def delete_state(
    state_id: str,
    client: SinasClient = Depends(sinas.require("sinas.contexts.delete:own"))
):
    await asyncio.to_thread(client.state.delete, state_id)
    return {"message": "Deleted"}

# Require ANY of multiple permissions (OR logic - default)
@app.post("/moderate-content")
async def moderate_content(
    client: SinasClient = Depends(
        sinas.require("sinas.admin:all", "sinas.moderation.write:all")
    )
):
    # User needs EITHER admin:all OR moderation.write:all
    return {"message": "Content moderated"}

# Require ALL permissions (AND logic)
@app.post("/admin-action")
async def admin_action(
    client: SinasClient = Depends(
        sinas.require("sinas.admin:all", "sinas.functions.write:all", logic="AND")
    )
):
    # User needs BOTH admin:all AND functions.write:all
    return {"message": "Admin action completed"}

# Alternative: Use require_all() for AND logic
@app.post("/super-admin-action")
async def super_admin_action(
    client: SinasClient = Depends(
        sinas.require_all("sinas.admin:all", "sinas.functions.write:all", "sinas.contexts.write:all")
    )
):
    # User needs ALL three permissions
    return {"message": "Super admin action completed"}

Permission Logic:

• OR logic (default): User needs at least one of the specified permissions
• AND logic: User needs all of the specified permissions
• Use logic="AND" parameter or require_all() method for AND logic

Complete Example

See examples/fastapi_app.py for a complete working example with both approaches.

Run it with:

uvicorn examples.fastapi_app:app --reload

Then visit http://localhost:8000/docs for interactive API documentation.

Context Manager

The client can be used as a context manager for proper resource cleanup:

with SinasClient(base_url="http://localhost:51245") as client:
    states = client.state.list()
    # Client is automatically closed when exiting

Error Handling

from sinas import (
    SinasClient,
    SinasAPIError,
    SinasAuthError,
    SinasNotFoundError,
    SinasValidationError
)

client = SinasClient(base_url="http://localhost:51245")

try:
    state = client.state.get("non-existent-id")
except SinasNotFoundError as e:
    print(f"State not found: {e}")
except SinasAuthError as e:
    print(f"Authentication failed: {e}")
except SinasValidationError as e:
    print(f"Validation error: {e}")
except SinasAPIError as e:
    print(f"API error: {e}")
    print(f"Status code: {e.status_code}")
    print(f"Response: {e.response}")

API Reference

SinasClient

auth

• login(email) - Send OTP to email
• verify_otp(session_id, otp_code) - Verify OTP and get tokens
• external_auth(token) - Exchange external OIDC token
• refresh(refresh_token) - Refresh access token
• logout(refresh_token) - Revoke refresh token
• get_me() - Get current user info
• check_permissions(permissions, logic="AND") - Check if user has permissions with AND/OR logic

state

• set(namespace, key, value, ...) - Create/update state
• get(state_id) - Get state by ID
• list(namespace?, visibility?, ...) - List states with filters
• update(state_id, value?, ...) - Update state
• delete(state_id) - Delete state

chats

• create(namespace, agent_name, input?, title?) - Create chat with agent
• send(chat_id, content) - Send message (blocking)
• stream(chat_id, content) - Send message (streaming SSE)
• get(chat_id) - Get chat with messages
• list() - List user's chats
• update(chat_id, title?) - Update chat
• delete(chat_id) - Delete chat

webhooks

• run(path, method?, body?, headers?, query?) - Execute webhook

executions

• list(function_name?, status?, skip?, limit?) - List executions
• get(execution_id) - Get execution details
• get_steps(execution_id) - Get execution steps
• continue_execution(execution_id, input) - Continue paused execution

FastAPI Integration

SinasAuth(base_url, auto_error=True)

FastAPI dependency for authentication and authorization.

Methods:

• require(*permissions, logic="OR") - Create dependency requiring permissions with AND/OR logic
• require_all(*permissions) - Create dependency requiring ALL permissions (AND logic)

Router Functions

• create_runtime_router(base_url, include_auth=False) - Complete runtime router
• create_state_router(base_url, prefix="") - State endpoints
• create_chat_router(base_url, prefix="") - Chat endpoints
• create_webhook_router(base_url, prefix="") - Webhook endpoints
• create_executions_router(base_url, prefix="") - Execution endpoints

Development

Install Development Dependencies

pip install -e ".[dev]"

Run Examples

# Basic usage
python examples/basic_usage.py

# FastAPI app
uvicorn examples.fastapi_app:app --reload

Code Formatting

black sinas
ruff check sinas

License

MIT