Python SDK

The ash-ai-sdk Python package provides a client for the Ash REST API. It is auto-generated from the OpenAPI specification.

Installation

Python

pip install ash-ai-sdk

TypeScript

npm install @ash-ai/sdk

Client Setup

Python

from ash_ai import AshClient

client = AshClient(
    server_url="http://localhost:4100",
    api_key="your-api-key",
)

TypeScript

import { AshClient } from '@ash-ai/sdk';

const client = new AshClient({
  serverUrl: 'http://localhost:4100',
  apiKey: 'your-api-key',
});

Usage Examples

Deploy an Agent

Python

agent = client.deploy_agent(name="my-agent", path="/path/to/agent")
print(f"Deployed: {agent.name} v{agent.version}")

TypeScript

const agent = await client.deployAgent('my-agent', '/path/to/agent');
console.log(`Deployed: ${agent.name} v${agent.version}`);

Create a Session

Python

session = client.create_session(agent="my-agent")
print(f"Session ID: {session.id}")
print(f"Status: {session.status}")

TypeScript

const session = await client.createSession('my-agent');
console.log(`Session ID: ${session.id}`);
console.log(`Status: ${session.status}`);

Send a Message (Streaming)

Python

for event in client.send_message_stream(session.id, "Analyze this code"):
    if event.type == "message":
        data = event.data
        if data.get("type") == "assistant" and data.get("message", {}).get("content"):
            for block in data["message"]["content"]:
                if block.get("type") == "text":
                    print(block["text"])
    elif event.type == "error":
        print(f"Error: {event.data['error']}")
    elif event.type == "done":
        print("Turn complete.")

TypeScript

for await (const event of client.sendMessageStream(session.id, 'Analyze this code')) {
  if (event.type === 'message') {
    const text = extractTextFromEvent(event.data);
    if (text) console.log(text);
  } else if (event.type === 'error') {
    console.error('Error:', event.data.error);
  } else if (event.type === 'done') {
    console.log('Turn complete.');
  }
}

Pause and Resume

Python

# Pause the session (persists workspace state)
paused = client.pause_session(session.id)
print(f"Status: {paused.status}")  # 'paused'

# Resume later (fast path if sandbox is still alive)
resumed = client.resume_session(session.id)
print(f"Status: {resumed.status}")  # 'active'

TypeScript

// Pause the session (persists workspace state)
const paused = await client.pauseSession(session.id);
console.log(`Status: ${paused.status}`); // 'paused'

// Resume later (fast path if sandbox is still alive)
const resumed = await client.resumeSession(session.id);
console.log(`Status: ${resumed.status}`); // 'active'

End a Session

Python

ended = client.end_session(session.id)
print(f"Status: {ended.status}")  # 'ended'

TypeScript

const ended = await client.endSession(session.id);
console.log(`Status: ${ended.status}`); // 'ended'

Multi-Turn Conversation

Python

session = client.create_session(agent="my-agent")

questions = [
    "What files are in the workspace?",
    "Read the main config file.",
    "Summarize what this project does.",
]

for question in questions:
    print(f"\n> {question}")
    for event in client.send_message_stream(session.id, question):
        if event.type == "message":
            data = event.data
            if data.get("type") == "assistant":
                content = data.get("message", {}).get("content", [])
                for block in content:
                    if block.get("type") == "text":
                        print(block["text"], end="")
    print()

client.end_session(session.id)

TypeScript

const session = await client.createSession('my-agent');

const questions = [
  'What files are in the workspace?',
  'Read the main config file.',
  'Summarize what this project does.',
];

for (const question of questions) {
  console.log(`\n> ${question}`);
  for await (const event of client.sendMessageStream(session.id, question)) {
    if (event.type === 'message') {
      const text = extractTextFromEvent(event.data);
      if (text) process.stdout.write(text);
    }
  }
  console.log();
}

await client.endSession(session.id);

List Agents and Sessions

Python

# List all deployed agents
agents = client.list_agents()
for agent in agents:
    print(f"{agent.name} (v{agent.version})")

# List all sessions, optionally filtered by agent
sessions = client.list_sessions(agent="my-agent")
for s in sessions:
    print(f"{s.id} - {s.status}")

TypeScript

// List all deployed agents
const agents = await client.listAgents();
for (const agent of agents) {
  console.log(`${agent.name} (v${agent.version})`);
}

// List all sessions, optionally filtered by agent
const sessions = await client.listSessions('my-agent');
for (const s of sessions) {
  console.log(`${s.id} - ${s.status}`);
}

Error Handling

Python

from ash_ai import AshApiError

try:
    session = client.create_session(agent="nonexistent")
except AshApiError as e:
    print(f"API error ({e.status_code}): {e.message}")
except Exception as e:
    print(f"Connection error: {e}")

TypeScript

try {
  const session = await client.createSession('nonexistent');
} catch (err) {
  console.error(err.message);
}

Note on SDK Generation

The Python SDK is auto-generated from the Ash server's OpenAPI specification using openapi-python-client. The spec is generated from Fastify route schemas, so the Python SDK always matches the server's API surface.

To regenerate the SDK from source:

make sdk-python

This runs make openapi first (to generate the spec), then runs the Python client generator.