Python Sdk

Python SDK

Build AI applications with the inference.sh Python SDK.

Quick Start

pip install inferencesh

from inferencesh import inference

client = inference(api_key="inf_your_key")

# Run an AI app
result = client.run({
    "app": "infsh/flux-1-dev",
    "input": {"prompt": "A sunset over mountains"}
})
print(result["output"])

Installation

# Standard installation
pip install inferencesh

# With async support
pip install inferencesh[async]

Requirements: Python 3.8+

Authentication

import os
from inferencesh import inference

# Direct API key
client = inference(api_key="inf_your_key")

# From environment variable (recommended)
client = inference(api_key=os.environ["INFERENCE_API_KEY"])

Get your API key: Settings → API Keys → Create API Key

Running Apps

Basic Execution

result = client.run({
    "app": "infsh/flux-1-dev",
    "input": {"prompt": "A cat astronaut"}
})

print(result["status"])  # "completed"
print(result["output"])  # Output data

Fire and Forget

task = client.run({
    "app": "google/veo-3-1-fast",
    "input": {"prompt": "Drone flying over mountains"}
}, wait=False)

print(f"Task ID: {task['id']}")
# Check later with client.get_task(task['id'])

Streaming Progress

for update in client.run({
    "app": "google/veo-3-1-fast",
    "input": {"prompt": "Ocean waves at sunset"}
}, stream=True):
    print(f"Status: {update['status']}")
    if update.get("logs"):
        print(update["logs"][-1])

Run Parameters

File Handling

Automatic Upload

result = client.run({
    "app": "image-processor",
    "input": {
        "image": "/path/to/image.png"  # Auto-uploaded
    }
})

Manual Upload

from inferencesh import UploadFileOptions

# Basic upload
file = client.upload_file("/path/to/image.png")

# With options
file = client.upload_file(
    "/path/to/image.png",
    UploadFileOptions(
        filename="custom_name.png",
        content_type="image/png",
        public=True
    )
)

result = client.run({
    "app": "image-processor",
    "input": {"image": file["uri"]}
})

Sessions (Stateful Execution)

Keep workers warm across multiple calls:

# Start new session
result = client.run({
    "app": "my-app",
    "input": {"action": "init"},
    "session": "new",
    "session_timeout": 300  # 5 minutes
})
session_id = result["session_id"]

# Continue in same session
result = client.run({
    "app": "my-app",
    "input": {"action": "process"},
    "session": session_id
})

Agent SDK

Template Agents

Use pre-built agents from your workspace:

agent = client.agent("my-team/support-agent@latest")

# Send message
response = agent.send_message("Hello!")
print(response.text)

# Multi-turn conversation
response = agent.send_message("Tell me more")

# Reset conversation
agent.reset()

# Get chat history
chat = agent.get_chat()

Ad-hoc Agents

Create custom agents programmatically:

from inferencesh import tool, string, number, app_tool

# Define tools
calculator = (
    tool("calculate")
    .describe("Perform a calculation")
    .param("expression", string("Math expression"))
    .build()
)

image_gen = (
    app_tool("generate_image", "infsh/flux-1-dev@latest")
    .describe("Generate an image")
    .param("prompt", string("Image description"))
    .build()
)

# Create agent
agent = client.agent({
    "core_app": {"ref": "infsh/claude-sonnet-4@latest"},
    "system_prompt": "You are a helpful assistant.",
    "tools": [calculator, image_gen],
    "temperature": 0.7,
    "max_tokens": 4096
})

response = agent.send_message("What is 25 * 4?")

Available Core Apps

Tool Builder API

Parameter Types

from inferencesh import (
    string, number, integer, boolean,
    enum_of, array, obj, optional
)

name = string("User's name")
age = integer("Age in years")
score = number("Score 0-1")
active = boolean("Is active")
priority = enum_of(["low", "medium", "high"], "Priority")
tags = array(string("Tag"), "List of tags")
address = obj({
    "street": string("Street"),
    "city": string("City"),
    "zip": optional(string("ZIP"))
}, "Address")

Client Tools (Run in Your Code)

greet = (
    tool("greet")
    .display("Greet User")
    .describe("Greets a user by name")
    .param("name", string("Name to greet"))
    .require_approval()
    .build()
)

App Tools (Call AI Apps)

generate = (
    app_tool("generate_image", "infsh/flux-1-dev@latest")
    .describe("Generate an image from text")
    .param("prompt", string("Image description"))
    .setup({"model": "schnell"})
    .input({"steps": 20})
    .require_approval()
    .build()
)

Agent Tools (Delegate to Sub-agents)

from inferencesh import agent_tool

researcher = (
    agent_tool("research", "my-org/researcher@v1")
    .describe("Research a topic")
    .param("topic", string("Topic to research"))
    .build()
)

Webhook Tools (Call External APIs)

from inferencesh import webhook_tool

notify = (
    webhook_tool("slack", "https://hooks.slack.com/...")
    .describe("Send Slack notification")
    .secret("SLACK_SECRET")
    .param("channel", string("Channel"))
    .param("message", string("Message"))
    .build()
)

Internal Tools (Built-in Capabilities)

from inferencesh import internal_tools

config = (
    internal_tools()
    .plan()
    .memory()
    .web_search(True)
    .code_execution(True)
    .image_generation({
        "enabled": True,
        "app_ref": "infsh/flux@latest"
    })
    .build()
)

agent = client.agent({
    "core_app": {"ref": "infsh/claude-sonnet-4@latest"},
    "internal_tools": config
})

Streaming Agent Responses

def handle_message(msg):
    if msg.get("content"):
        print(msg["content"], end="", flush=True)

def handle_tool(call):
    print(f"\n[Tool: {call.name}]")
    result = execute_tool(call.name, call.args)
    agent.submit_tool_result(call.id, result)

response = agent.send_message(
    "Explain quantum computing",
    on_message=handle_message,
    on_tool_call=handle_tool
)

File Attachments

# From file path
with open("image.png", "rb") as f:
    response = agent.send_message(
        "What's in this image?",
        files=[f.read()]
    )

# From base64
response = agent.send_message(
    "Analyze this",
    files=["data:image/png;base64,iVBORw0KGgo..."]
)

Skills (Reusable Context)

agent = client.agent({
    "core_app": {"ref": "infsh/claude-sonnet-4@latest"},
    "skills": [
        {
            "name": "code-review",
            "description": "Code review guidelines",
            "content": "# Code Review\n\n1. Check security\n2. Check performance..."
        },
        {
            "name": "api-docs",
            "description": "API documentation",
            "url": "https://example.com/skills/api-docs.md"
        }
    ]
})

Async Support

from inferencesh import async_inference
import asyncio

async def main():
    client = async_inference(api_key="inf_...")

    # Async app execution
    result = await client.run({
        "app": "infsh/flux-1-dev",
        "input": {"prompt": "A galaxy"}
    })

    # Async agent
    agent = client.agent("my-org/assistant@latest")
    response = await agent.send_message("Hello!")

    # Async streaming
    async for msg in agent.stream_messages():
        print(msg)

asyncio.run(main())

Error Handling

from inferencesh import RequirementsNotMetException

try:
    result = client.run({"app": "my-app", "input": {...}})
except RequirementsNotMetException as e:
    print(f"Missing requirements:")
    for err in e.errors:
        print(f"  - {err['type']}: {err['key']}")
except RuntimeError as e:
    print(f"Error: {e}")

Human Approval Workflows

def handle_tool(call):
    if call.requires_approval:
        # Show to user, get confirmation
        approved = prompt_user(f"Allow {call.name}?")
        if approved:
            result = execute_tool(call.name, call.args)
            agent.submit_tool_result(call.id, result)
        else:
            agent.submit_tool_result(call.id, {"error": "Denied by user"})

response = agent.send_message(
    "Delete all temp files",
    on_tool_call=handle_tool
)

Reference Files

• Agent Patterns - Multi-agent, RAG, human-in-the-loop patterns
• Tool Builder - Complete tool builder API reference
• Streaming - Real-time progress updates and SSE handling
• File Handling - Upload, download, and manage files
• Sessions - Stateful execution with warm workers
• Async Patterns - Parallel processing and async/await

Related Skills

# JavaScript SDK
npx skills add inference-sh/skills@javascript-sdk

# Full platform skill (all 150+ apps via CLI)
npx skills add inference-sh/skills@infsh-cli

# LLM models
npx skills add inference-sh/skills@llm-models

# Image generation
npx skills add inference-sh/skills@ai-image-generation

Documentation

• Python SDK Reference - Full API documentation
• Agent SDK Overview - Building agents
• Tool Builder Reference - Creating tools
• Authentication - API key setup
• Streaming - Real-time updates
• File Uploads - File handling

Agent Patterns

Common patterns for building agents with the Python SDK.

Multi-Agent Orchestration

Delegate tasks to specialized sub-agents:

from inferencesh import inference, agent_tool, string

client = inference(api_key="inf_...")

# Define sub-agents as tools
researcher = (
    agent_tool("research", "my-org/researcher@latest")
    .describe("Research a topic thoroughly")
    .param("topic", string("Topic to research"))
    .build()
)

writer = (
    agent_tool("write", "my-org/writer@latest")
    .describe("Write content based on research")
    .param("outline", string("Content outline"))
    .param("research", string("Research findings"))
    .build()
)

# Orchestrator agent
orchestrator = client.agent({
    "core_app": {"ref": "infsh/claude-sonnet-4@latest"},
    "system_prompt": """You are an orchestrator that:
1. Uses the research tool to gather information
2. Uses the write tool to create content
Coordinate between agents to produce high-quality output.""",
    "tools": [researcher, writer]
})

response = orchestrator.send_message("Create a blog post about AI agents")

RAG Pattern (Retrieval-Augmented Generation)

Combine search with LLM responses:

from inferencesh import inference, app_tool, string

client = inference(api_key="inf_...")

# Search tool
search = (
    app_tool("search", "tavily/search-assistant@latest")
    .describe("Search the web for current information")
    .param("query", string("Search query"))
    .build()
)

# RAG agent
rag_agent = client.agent({
    "core_app": {"ref": "infsh/claude-sonnet-4@latest"},
    "system_prompt": """You help users with current information.
When asked about recent events or facts you're unsure about,
use the search tool to find accurate, up-to-date information.
Always cite your sources.""",
    "tools": [search]
})

response = rag_agent.send_message("What are the latest developments in quantum computing?")

Code Execution Pattern

Agents that can write and run code:

from inferencesh import inference, internal_tools

client = inference(api_key="inf_...")

config = (
    internal_tools()
    .code_execution(True)
    .build()
)

coder = client.agent({
    "core_app": {"ref": "infsh/claude-sonnet-4@latest"},
    "system_prompt": """You are a Python coding assistant.
Write code to solve problems and execute it to verify it works.
Explain your approach and show the output.""",
    "internal_tools": config
})

response = coder.send_message("Calculate the first 20 Fibonacci numbers")

Human-in-the-Loop Pattern

Require approval for sensitive operations:

from inferencesh import inference, tool, string

client = inference(api_key="inf_...")

# Tool requiring approval
delete_file = (
    tool("delete_file")
    .describe("Delete a file from the filesystem")
    .param("path", string("File path to delete"))
    .require_approval()
    .build()
)

def handle_tool(call):
    if call.requires_approval:
        print(f"\n⚠️  Agent wants to: {call.name}")
        print(f"   Arguments: {call.args}")
        confirm = input("Allow? (y/n): ")

        if confirm.lower() == 'y':
            result = execute_operation(call.name, call.args)
            agent.submit_tool_result(call.id, result)
        else:
            agent.submit_tool_result(call.id, {
                "error": "Operation denied by user"
            })
    else:
        result = execute_operation(call.name, call.args)
        agent.submit_tool_result(call.id, result)

agent = client.agent({
    "core_app": {"ref": "infsh/claude-sonnet-4@latest"},
    "tools": [delete_file]
})

response = agent.send_message(
    "Clean up temporary files in /tmp/myapp",
    on_tool_call=handle_tool
)

Conversation Memory Pattern

Maintain context across sessions:

import json
from inferencesh import inference

client = inference(api_key="inf_...")

def save_chat(agent, filepath):
    chat = agent.get_chat()
    with open(filepath, 'w') as f:
        json.dump(chat, f)

def load_chat(agent, filepath):
    try:
        with open(filepath, 'r') as f:
            chat = json.load(f)
            # Restore conversation by replaying messages
            for msg in chat['messages']:
                if msg['role'] == 'user':
                    agent.send_message(msg['content'])
    except FileNotFoundError:
        pass

agent = client.agent("my-org/assistant@latest")

# Load previous conversation
load_chat(agent, "conversation.json")

# Continue conversation
response = agent.send_message("Continue where we left off")

# Save for next session
save_chat(agent, "conversation.json")

Streaming with Progress UI

Real-time updates for better UX:

from inferencesh import inference
import sys

client = inference(api_key="inf_...")
agent = client.agent("my-org/assistant@latest")

def stream_handler(msg):
    if msg.get("content"):
        sys.stdout.write(msg["content"])
        sys.stdout.flush()

def tool_handler(call):
    print(f"\n🔧 Using tool: {call.name}")
    # Execute and return result
    result = execute_tool(call.name, call.args)
    agent.submit_tool_result(call.id, result)
    print("✅ Tool completed")

response = agent.send_message(
    "Generate a report on market trends",
    on_message=stream_handler,
    on_tool_call=tool_handler
)

print("\n\n📊 Report complete!")

Error Recovery Pattern

Graceful handling of failures:

from inferencesh import inference, RequirementsNotMetException
import time

client = inference(api_key="inf_...")

def robust_run(config, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.run(config)
        except RequirementsNotMetException as e:
            print(f"Missing requirements: {e.errors}")
            raise
        except RuntimeError as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt
                print(f"Error: {e}. Retrying in {wait}s...")
                time.sleep(wait)
            else:
                raise

result = robust_run({
    "app": "infsh/flux-1-dev",
    "input": {"prompt": "A serene landscape"}
})

Batch Processing Pattern

Process multiple items efficiently:

from inferencesh import async_inference
import asyncio

async def process_batch(items):
    client = async_inference(api_key="inf_...")

    async def process_one(item):
        result = await client.run({
            "app": "infsh/flux-1-dev",
            "input": {"prompt": item}
        })
        return result

    # Process in parallel with concurrency limit
    semaphore = asyncio.Semaphore(5)  # Max 5 concurrent

    async def bounded_process(item):
        async with semaphore:
            return await process_one(item)

    results = await asyncio.gather(*[
        bounded_process(item) for item in items
    ])
    return results

prompts = [
    "A mountain sunrise",
    "A city at night",
    "An ocean sunset",
    "A forest path"
]

results = asyncio.run(process_batch(prompts))

Async Patterns Reference

Asynchronous programming with the Python SDK.

Basic Async Client

from inferencesh import async_inference
import asyncio

async def main():
    client = async_inference(api_key="inf_...")

    result = await client.run({
        "app": "infsh/flux-1-dev",
        "input": {"prompt": "A sunset"}
    })
    print(result["output"])

asyncio.run(main())

Parallel Requests

Run multiple independent requests simultaneously:

async def parallel_requests():
    client = async_inference(api_key="inf_...")

    prompts = [
        "A mountain landscape",
        "An ocean sunset",
        "A forest path",
        "A city skyline"
    ]

    # Create all tasks
    tasks = [
        client.run({
            "app": "infsh/flux-1-dev",
            "input": {"prompt": p}
        })
        for p in prompts
    ]

    # Run in parallel
    results = await asyncio.gather(*tasks)

    return results

results = asyncio.run(parallel_requests())

Controlled Concurrency

Limit concurrent requests with semaphore:

async def controlled_concurrency(items, max_concurrent=5):
    client = async_inference(api_key="inf_...")
    semaphore = asyncio.Semaphore(max_concurrent)

    async def process_one(item):
        async with semaphore:
            return await client.run({
                "app": "processor",
                "input": {"data": item}
            })

    tasks = [process_one(item) for item in items]
    return await asyncio.gather(*tasks)

# Process 100 items, max 5 at a time
results = asyncio.run(controlled_concurrency(range(100), max_concurrent=5))

Async Streaming

async def stream_task():
    client = async_inference(api_key="inf_...")

    async for update in client.run({
        "app": "google/veo-3-1-fast",
        "input": {"prompt": "Ocean waves"}
    }, stream=True):
        print(f"Status: {update['status']}")

        if update.get("status") == "completed":
            return update.get("output")

result = asyncio.run(stream_task())

Async Agents

async def agent_conversation():
    client = async_inference(api_key="inf_...")
    agent = client.agent("my-org/assistant@latest")

    # Send message
    response = await agent.send_message("Hello!")
    print(response.text)

    # Streaming with callback
    async def on_message(msg):
        if msg.get("content"):
            print(msg["content"], end="", flush=True)

    response = await agent.send_message(
        "Tell me a story",
        on_message=on_message
    )

asyncio.run(agent_conversation())

Producer-Consumer Pattern

async def producer_consumer():
    client = async_inference(api_key="inf_...")
    queue = asyncio.Queue()
    results = []

    async def producer(items):
        for item in items:
            await queue.put(item)
        # Signal end
        for _ in range(3):  # Number of consumers
            await queue.put(None)

    async def consumer(consumer_id):
        while True:
            item = await queue.get()
            if item is None:
                break

            result = await client.run({
                "app": "processor",
                "input": {"data": item}
            })
            results.append((item, result))
            print(f"Consumer {consumer_id} processed {item}")

    items = list(range(20))

    # Start producer and consumers
    await asyncio.gather(
        producer(items),
        consumer(1),
        consumer(2),
        consumer(3)
    )

    return results

results = asyncio.run(producer_consumer())

Timeout Handling

async def with_timeout():
    client = async_inference(api_key="inf_...")

    try:
        result = await asyncio.wait_for(
            client.run({
                "app": "slow-app",
                "input": {"data": "..."}
            }),
            timeout=30.0  # 30 seconds
        )
        return result
    except asyncio.TimeoutError:
        print("Request timed out")
        return None

result = asyncio.run(with_timeout())

Retry with Backoff

async def retry_with_backoff(client, config, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await client.run(config)
        except Exception as e:
            if attempt == max_retries - 1:
                raise

            wait = (2 ** attempt) + random.random()
            print(f"Attempt {attempt + 1} failed, retrying in {wait:.1f}s...")
            await asyncio.sleep(wait)

async def main():
    client = async_inference(api_key="inf_...")

    result = await retry_with_backoff(client, {
        "app": "my-app",
        "input": {"data": "..."}
    })

asyncio.run(main())

Batch Processing with Progress

from tqdm.asyncio import tqdm

async def batch_with_progress(items):
    client = async_inference(api_key="inf_...")
    semaphore = asyncio.Semaphore(10)

    async def process_one(item):
        async with semaphore:
            return await client.run({
                "app": "processor",
                "input": {"data": item}
            })

    tasks = [process_one(item) for item in items]

    results = []
    for coro in tqdm.as_completed(tasks, desc="Processing"):
        result = await coro
        results.append(result)

    return results

results = asyncio.run(batch_with_progress(range(100)))

Context Manager Pattern

class AsyncInferenceSession:
    def __init__(self, api_key):
        self.api_key = api_key
        self.client = None

    async def __aenter__(self):
        self.client = async_inference(api_key=self.api_key)
        return self.client

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # Cleanup if needed
        pass

async def main():
    async with AsyncInferenceSession("inf_...") as client:
        result = await client.run({
            "app": "my-app",
            "input": {"data": "..."}
        })
        print(result)

asyncio.run(main())

Error Aggregation

async def process_with_errors(items):
    client = async_inference(api_key="inf_...")

    async def safe_process(item):
        try:
            result = await client.run({
                "app": "processor",
                "input": {"data": item}
            })
            return {"success": True, "item": item, "result": result}
        except Exception as e:
            return {"success": False, "item": item, "error": str(e)}

    tasks = [safe_process(item) for item in items]
    results = await asyncio.gather(*tasks)

    successes = [r for r in results if r["success"]]
    failures = [r for r in results if not r["success"]]

    print(f"Succeeded: {len(successes)}, Failed: {len(failures)}")

    return successes, failures

asyncio.run(process_with_errors(range(50)))

Async Generator Pattern

async def stream_results(items):
    """Yield results as they complete."""
    client = async_inference(api_key="inf_...")

    pending = set()

    for item in items:
        task = asyncio.create_task(client.run({
            "app": "processor",
            "input": {"data": item}
        }))
        task.item = item
        pending.add(task)

        # Limit pending tasks
        if len(pending) >= 10:
            done, pending = await asyncio.wait(
                pending,
                return_when=asyncio.FIRST_COMPLETED
            )
            for task in done:
                yield task.item, await task

    # Wait for remaining
    while pending:
        done, pending = await asyncio.wait(
            pending,
            return_when=asyncio.FIRST_COMPLETED
        )
        for task in done:
            yield task.item, await task

async def main():
    async for item, result in stream_results(range(100)):
        print(f"Completed: {item}")

asyncio.run(main())

Integration with Web Frameworks

FastAPI

from fastapi import FastAPI
from inferencesh import async_inference

app = FastAPI()
client = async_inference(api_key="inf_...")

@app.post("/generate")
async def generate(prompt: str):
    result = await client.run({
        "app": "infsh/flux-1-dev",
        "input": {"prompt": prompt}
    })
    return {"image": result["output"]["url"]}

aiohttp

from aiohttp import web
from inferencesh import async_inference

client = async_inference(api_key="inf_...")

async def handle_generate(request):
    data = await request.json()
    result = await client.run({
        "app": "infsh/flux-1-dev",
        "input": {"prompt": data["prompt"]}
    })
    return web.json_response({"image": result["output"]["url"]})

app = web.Application()
app.router.add_post('/generate', handle_generate)

File Handling Reference

Upload, download, and manage files with the Python SDK.

Automatic File Upload

Local file paths in input are automatically uploaded:

from inferencesh import inference

client = inference(api_key="inf_...")

# File path is auto-uploaded
result = client.run({
    "app": "image-processor",
    "input": {
        "image": "/path/to/image.png"
    }
})

Manual File Upload

Basic Upload

file = client.upload_file("/path/to/image.png")
print(file["uri"])  # inf://files/abc123

result = client.run({
    "app": "image-processor",
    "input": {"image": file["uri"]}
})

Upload Options

from inferencesh import UploadFileOptions

file = client.upload_file(
    "/path/to/document.pdf",
    UploadFileOptions(
        filename="custom_name.pdf",      # Custom filename
        content_type="application/pdf",   # MIME type
        path="/documents/reports",        # Storage path
        public=True                       # Publicly accessible
    )
)

Supported Input Types

File Path

result = client.run({
    "app": "processor",
    "input": {"file": "/path/to/file.png"}
})

Data URI (Base64)

import base64

with open("image.png", "rb") as f:
    b64 = base64.b64encode(f.read()).decode()

result = client.run({
    "app": "processor",
    "input": {"image": f"data:image/png;base64,{b64}"}
})

Bytes

with open("image.png", "rb") as f:
    data = f.read()

file = client.upload_file(data, UploadFileOptions(
    filename="image.png",
    content_type="image/png"
))

File Object

with open("image.png", "rb") as f:
    file = client.upload_file(f, UploadFileOptions(
        filename="image.png",
        content_type="image/png"
    ))

Working with URLs

Use remote URLs directly (no upload needed):

result = client.run({
    "app": "image-processor",
    "input": {
        "image": "https://example.com/image.png"
    }
})

Multiple Files

# Upload multiple files
files = []
for path in ["/path/to/file1.png", "/path/to/file2.png"]:
    file = client.upload_file(path)
    files.append(file["uri"])

result = client.run({
    "app": "multi-file-processor",
    "input": {"images": files}
})

File Info

file = client.upload_file("/path/to/image.png")

print(f"URI: {file['uri']}")
print(f"URL: {file['url']}")  # Direct access URL
print(f"Size: {file['size']}")
print(f"Type: {file['content_type']}")

Downloading Results

import requests

result = client.run({
    "app": "infsh/flux-1-dev",
    "input": {"prompt": "A sunset"}
})

# Result contains URL to generated file
image_url = result["output"]["image"]

# Download the file
response = requests.get(image_url)
with open("output.png", "wb") as f:
    f.write(response.content)

Async File Operations

from inferencesh import async_inference
import asyncio
import aiohttp

async def process_files():
    client = async_inference(api_key="inf_...")

    # Upload
    file = await client.upload_file("/path/to/image.png")

    # Process
    result = await client.run({
        "app": "image-processor",
        "input": {"image": file["uri"]}
    })

    # Download result
    async with aiohttp.ClientSession() as session:
        async with session.get(result["output"]["url"]) as resp:
            data = await resp.read()
            with open("output.png", "wb") as f:
                f.write(data)

asyncio.run(process_files())

Agent File Attachments

agent = client.agent("my-org/assistant@latest")

# From bytes
with open("image.png", "rb") as f:
    response = agent.send_message(
        "What's in this image?",
        files=[f.read()]
    )

# From base64
response = agent.send_message(
    "Analyze this document",
    files=["data:application/pdf;base64,JVBERi0xLj..."]
)

# Multiple files
with open("img1.png", "rb") as f1, open("img2.png", "rb") as f2:
    response = agent.send_message(
        "Compare these images",
        files=[f1.read(), f2.read()]
    )

Large File Handling

For large files, use chunked upload:

def upload_large_file(client, filepath, chunk_size=5*1024*1024):
    """Upload large file in chunks (5MB default)."""
    import os

    file_size = os.path.getsize(filepath)
    filename = os.path.basename(filepath)

    with open(filepath, 'rb') as f:
        # Initialize multipart upload
        upload = client.create_multipart_upload(
            filename=filename,
            content_type="application/octet-stream",
            size=file_size
        )

        parts = []
        part_number = 1

        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break

            part = client.upload_part(
                upload_id=upload["id"],
                part_number=part_number,
                data=chunk
            )
            parts.append(part)
            part_number += 1

        # Complete upload
        file = client.complete_multipart_upload(
            upload_id=upload["id"],
            parts=parts
        )

        return file

Content Type Detection

import mimetypes

def upload_with_auto_type(client, filepath):
    content_type, _ = mimetypes.guess_type(filepath)

    return client.upload_file(
        filepath,
        UploadFileOptions(
            content_type=content_type or "application/octet-stream"
        )
    )

Temporary Files

import tempfile
import os

# Create temp file for processing
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp:
    tmp.write(image_data)
    tmp_path = tmp.name

try:
    result = client.run({
        "app": "image-processor",
        "input": {"image": tmp_path}
    })
finally:
    os.unlink(tmp_path)  # Clean up

Error Handling

from inferencesh import FileUploadError

try:
    file = client.upload_file("/path/to/large_file.bin")
except FileUploadError as e:
    if "too large" in str(e):
        print("File exceeds size limit")
    elif "unsupported" in str(e):
        print("File type not supported")
    else:
        print(f"Upload failed: {e}")

Sessions Reference

Stateful execution with warm workers.

What Are Sessions?

Sessions keep workers warm between requests, enabling:

• Faster execution - No cold start on subsequent calls
• Shared state - Maintain context, loaded models, cached data
• Cost efficiency - Reuse initialized resources

Creating a Session

from inferencesh import inference

client = inference(api_key="inf_...")

# Start new session
result = client.run({
    "app": "my-app",
    "input": {"action": "initialize"},
    "session": "new"
})

session_id = result["session_id"]
print(f"Session: {session_id}")

Using an Existing Session

# Continue in same session
result = client.run({
    "app": "my-app",
    "input": {"action": "process", "data": "..."},
    "session": session_id
})

Session Timeout

Set how long idle sessions stay alive (1-3600 seconds):

# 5-minute timeout
result = client.run({
    "app": "my-app",
    "input": {"action": "init"},
    "session": "new",
    "session_timeout": 300
})

Session Lifecycle

1. Create session (session: "new")
   ↓
2. Worker starts, initializes app
   ↓
3. Subsequent calls reuse worker (session: session_id)
   ↓
4. Idle timeout reached or explicit close
   ↓
5. Worker terminates

Use Cases

Model Loading

Load a model once, use it multiple times:

# Initial load (slow)
result = client.run({
    "app": "ml-inference",
    "input": {"action": "load_model", "model": "large-model-v2"},
    "session": "new",
    "session_timeout": 600
})
session_id = result["session_id"]

# Fast inference calls
for item in data_batch:
    result = client.run({
        "app": "ml-inference",
        "input": {"action": "predict", "data": item},
        "session": session_id
    })
    print(result["output"])

Browser Automation

Keep browser open across multiple actions:

# Start browser session
result = client.run({
    "app": "browser-automation",
    "input": {"action": "start", "url": "https://example.com"},
    "session": "new",
    "session_timeout": 300
})
session_id = result["session_id"]

# Navigate
client.run({
    "app": "browser-automation",
    "input": {"action": "click", "selector": "#login-btn"},
    "session": session_id
})

# Fill form
client.run({
    "app": "browser-automation",
    "input": {"action": "type", "selector": "#username", "text": "user@example.com"},
    "session": session_id
})

# Take screenshot
result = client.run({
    "app": "browser-automation",
    "input": {"action": "screenshot"},
    "session": session_id
})

Stateful Conversations

# Initialize chat context
result = client.run({
    "app": "chat-with-memory",
    "input": {"action": "init", "system": "You are a helpful assistant."},
    "session": "new",
    "session_timeout": 1800  # 30 minutes
})
session_id = result["session_id"]

# Multi-turn conversation
messages = [
    "What is quantum computing?",
    "Can you give me a simple example?",
    "How is it different from classical computing?"
]

for msg in messages:
    result = client.run({
        "app": "chat-with-memory",
        "input": {"message": msg},
        "session": session_id
    })
    print(f"Assistant: {result['output']['response']}")

Data Processing Pipeline

# Load data once
result = client.run({
    "app": "data-processor",
    "input": {"action": "load", "dataset": "large_dataset.parquet"},
    "session": "new",
    "session_timeout": 900
})
session_id = result["session_id"]

# Run multiple analyses
analyses = ["summary", "correlations", "outliers", "trends"]

for analysis in analyses:
    result = client.run({
        "app": "data-processor",
        "input": {"action": "analyze", "type": analysis},
        "session": session_id
    })
    print(f"{analysis}: {result['output']}")

Session Management

Check Session Status

# Sessions are implicitly active when used
# If session expired, you'll get an error
try:
    result = client.run({
        "app": "my-app",
        "input": {"action": "check"},
        "session": session_id
    })
except Exception as e:
    if "session not found" in str(e).lower():
        print("Session expired, creating new one")
        # Create new session

Explicit Session Close

# Close session to free resources
client.run({
    "app": "my-app",
    "input": {"action": "cleanup"},
    "session": session_id
})
# Session will terminate after this call

Session Recovery Pattern

class SessionManager:
    def __init__(self, client, app, timeout=300):
        self.client = client
        self.app = app
        self.timeout = timeout
        self.session_id = None

    def ensure_session(self):
        if self.session_id is None:
            result = self.client.run({
                "app": self.app,
                "input": {"action": "init"},
                "session": "new",
                "session_timeout": self.timeout
            })
            self.session_id = result["session_id"]
        return self.session_id

    def run(self, input_data):
        try:
            return self.client.run({
                "app": self.app,
                "input": input_data,
                "session": self.ensure_session()
            })
        except Exception as e:
            if "session" in str(e).lower():
                # Session expired, create new one
                self.session_id = None
                return self.client.run({
                    "app": self.app,
                    "input": input_data,
                    "session": self.ensure_session()
                })
            raise

# Usage
manager = SessionManager(client, "my-app", timeout=600)
result = manager.run({"action": "process", "data": "..."})

Async Sessions

from inferencesh import async_inference
import asyncio

async def session_workflow():
    client = async_inference(api_key="inf_...")

    # Create session
    result = await client.run({
        "app": "my-app",
        "input": {"action": "init"},
        "session": "new",
        "session_timeout": 300
    })
    session_id = result["session_id"]

    # Run operations
    tasks = [
        client.run({
            "app": "my-app",
            "input": {"action": "process", "id": i},
            "session": session_id
        })
        for i in range(10)
    ]

    # Note: These run sequentially on the same worker
    results = []
    for task in tasks:
        results.append(await task)

    return results

asyncio.run(session_workflow())

Best Practices

1. Set appropriate timeouts - Balance between keeping workers warm and resource usage 2. Handle session expiry - Always catch and handle session not found errors 3. Clean up when done - Close sessions explicitly if you know you're finished 4. Don't over-parallelize - Session requests go to the same worker sequentially 5. Monitor costs - Long-running sessions incur ongoing charges

Streaming Reference

Real-time progress updates and Server-Sent Events (SSE) handling.

Task Status Flow

RECEIVED (1) → QUEUED (2) → SCHEDULED (3) → PREPARING (4)
→ SERVING (5) → SETTING_UP (6) → RUNNING (7) → UPLOADING (8)
→ COMPLETED (10), FAILED (11), or CANCELLED (12)

Basic Streaming

from inferencesh import inference

client = inference(api_key="inf_...")

for update in client.run({
    "app": "google/veo-3-1-fast",
    "input": {"prompt": "A sunset timelapse"}
}, stream=True):
    print(f"Status: {update['status']}")

Handling Different Update Types

for update in client.run(config, stream=True):
    status = update.get("status")

    # Task state changes
    if status == "queued":
        print("Task queued, waiting for worker...")
    elif status == "running":
        print("Task is running...")
    elif status == "completed":
        print("Done!")
        print(f"Output: {update.get('output')}")
    elif status == "failed":
        print(f"Error: {update.get('error')}")

    # Progress logs
    if update.get("logs"):
        for log in update["logs"]:
            print(f"  Log: {log}")

    # Partial outputs
    if update.get("partial_output"):
        print(f"  Partial: {update['partial_output']}")

Progress Tracking with UI

import sys

def progress_bar(current, total, width=50):
    filled = int(width * current / total)
    bar = "█" * filled + "░" * (width - filled)
    percent = current / total * 100
    sys.stdout.write(f"\r[{bar}] {percent:.1f}%")
    sys.stdout.flush()

for update in client.run(config, stream=True):
    if update.get("progress"):
        progress_bar(update["progress"]["current"], update["progress"]["total"])

    if update.get("status") == "completed":
        print("\n✓ Complete!")

Streaming with Timeout

import time

start = time.time()
timeout = 300  # 5 minutes

for update in client.run(config, stream=True):
    if time.time() - start > timeout:
        print("Timeout reached")
        break

    print(f"Status: {update['status']}")

    if update.get("status") in ["completed", "failed"]:
        break

Async Streaming

from inferencesh import async_inference
import asyncio

async def stream_task():
    client = async_inference(api_key="inf_...")

    async for update in client.run({
        "app": "google/veo-3-1-fast",
        "input": {"prompt": "Ocean waves"}
    }, stream=True):
        print(f"Status: {update['status']}")

        if update.get("status") == "completed":
            return update.get("output")

result = asyncio.run(stream_task())

Agent Streaming

agent = client.agent("my-org/assistant@latest")

def on_message(msg):
    if msg.get("content"):
        # Stream text as it arrives
        print(msg["content"], end="", flush=True)

    if msg.get("type") == "thinking":
        print(f"\n[Thinking: {msg.get('content')}]")

def on_tool_call(call):
    print(f"\n[Calling tool: {call.name}]")
    result = execute_tool(call.name, call.args)
    agent.submit_tool_result(call.id, result)

response = agent.send_message(
    "Explain quantum entanglement",
    on_message=on_message,
    on_tool_call=on_tool_call
)

Reconnection Handling

from inferencesh import inference, StreamingOptions

client = inference(api_key="inf_...")

options = StreamingOptions(
    max_retries=3,
    retry_delay=1.0,  # seconds
    chunk_size=1024
)

for update in client.run(config, stream=True, options=options):
    print(update)

Multiple Streams in Parallel

from inferencesh import async_inference
import asyncio

async def run_parallel():
    client = async_inference(api_key="inf_...")

    configs = [
        {"app": "infsh/flux-1-dev", "input": {"prompt": "A mountain"}},
        {"app": "infsh/flux-1-dev", "input": {"prompt": "An ocean"}},
        {"app": "infsh/flux-1-dev", "input": {"prompt": "A forest"}}
    ]

    async def stream_one(config, index):
        async for update in client.run(config, stream=True):
            print(f"[{index}] {update['status']}")
            if update.get("status") == "completed":
                return update.get("output")

    results = await asyncio.gather(*[
        stream_one(c, i) for i, c in enumerate(configs)
    ])
    return results

results = asyncio.run(run_parallel())

Cancelling a Stream

task_id = None

try:
    for update in client.run(config, stream=True):
        task_id = update.get("id")
        print(f"Status: {update['status']}")

        if should_cancel():
            break
finally:
    if task_id:
        client.cancel_task(task_id)
        print("Task cancelled")

Collecting All Logs

all_logs = []

for update in client.run(config, stream=True):
    if update.get("logs"):
        all_logs.extend(update["logs"])

    if update.get("status") == "completed":
        print("Final logs:")
        for log in all_logs:
            print(f"  {log}")

Custom Stream Processing

class StreamProcessor:
    def __init__(self):
        self.logs = []
        self.start_time = None
        self.end_time = None

    def process(self, update):
        if self.start_time is None:
            self.start_time = time.time()

        if update.get("logs"):
            self.logs.extend(update["logs"])

        if update.get("status") in ["completed", "failed"]:
            self.end_time = time.time()
            return True  # Done

        return False  # Continue

    @property
    def duration(self):
        if self.start_time and self.end_time:
            return self.end_time - self.start_time
        return None

processor = StreamProcessor()

for update in client.run(config, stream=True):
    if processor.process(update):
        break

print(f"Duration: {processor.duration:.2f}s")
print(f"Logs: {len(processor.logs)}")

Tool Builder Reference

Complete guide to building tools with the Python SDK.

Parameter Types

Basic Types

from inferencesh import string, number, integer, boolean

# String parameter
name = string("The user's full name")

# Number (float)
score = number("Score between 0 and 1")

# Integer
count = integer("Number of items")

# Boolean
enabled = boolean("Whether feature is enabled")

Enum Type

from inferencesh import enum_of

priority = enum_of(
    ["low", "medium", "high", "critical"],
    "Task priority level"
)

Array Type

from inferencesh import array, string

# Array of strings
tags = array(string("Tag name"), "List of tags")

# Array of objects
items = array(
    obj({
        "name": string("Item name"),
        "qty": integer("Quantity")
    }),
    "List of items"
)

Object Type

from inferencesh import obj, string, integer, optional

address = obj({
    "street": string("Street address"),
    "city": string("City name"),
    "state": string("State code"),
    "zip": optional(string("ZIP code"))
}, "Mailing address")

Optional Parameters

from inferencesh import optional, string

# Optional string
nickname = optional(string("User's nickname"))

Tool Types

Client Tools

Tools that execute in your code:

from inferencesh import tool, string, integer

# Basic tool
greet = (
    tool("greet")
    .describe("Greets a user")
    .param("name", string("Name to greet"))
    .build()
)

# Tool with multiple parameters
send_email = (
    tool("send_email")
    .display("Send Email")
    .describe("Sends an email to a recipient")
    .param("to", string("Recipient email"))
    .param("subject", string("Email subject"))
    .param("body", string("Email body"))
    .param("priority", integer("Priority 1-5"), default=3)
    .require_approval()
    .build()
)

App Tools

Tools that call inference.sh apps:

from inferencesh import app_tool, string

# Basic app tool
generate = (
    app_tool("generate_image", "infsh/flux-schnell@latest")
    .describe("Generate an image from a text prompt")
    .param("prompt", string("Image description"))
    .build()
)

# App tool with setup and defaults
translate = (
    app_tool("translate", "infsh/translator@latest")
    .describe("Translate text between languages")
    .param("text", string("Text to translate"))
    .param("target_lang", string("Target language code"))
    .setup({
        "model": "advanced",
        "preserve_formatting": True
    })
    .input({
        "source_lang": "auto"
    })
    .build()
)

Agent Tools

Tools that delegate to other agents:

from inferencesh import agent_tool, string

researcher = (
    agent_tool("research", "my-org/researcher@v1")
    .describe("Research a topic in depth")
    .param("topic", string("Topic to research"))
    .param("depth", string("Research depth: brief, moderate, comprehensive"))
    .build()
)

coder = (
    agent_tool("write_code", "my-org/coder@latest")
    .describe("Write code to solve a problem")
    .param("task", string("Coding task description"))
    .param("language", string("Programming language"))
    .build()
)

Webhook Tools

Tools that call external HTTP endpoints:

from inferencesh import webhook_tool, string

# Slack notification
slack = (
    webhook_tool("notify_slack", "https://hooks.slack.com/services/...")
    .describe("Send a message to Slack")
    .param("channel", string("Channel name"))
    .param("message", string("Message text"))
    .build()
)

# Webhook with secret
github = (
    webhook_tool("create_issue", "https://api.github.com/repos/org/repo/issues")
    .describe("Create a GitHub issue")
    .secret("GITHUB_TOKEN")  # Uses stored secret
    .param("title", string("Issue title"))
    .param("body", string("Issue description"))
    .build()
)

Tool Builder Methods

Common Methods

App Tool Methods

Webhook Tool Methods

Internal Tools

Built-in capabilities you can enable:

from inferencesh import internal_tools

config = (
    internal_tools()
    .plan()                    # Task planning
    .memory()                  # Information storage
    .web_search(True)         # Web search capability
    .code_execution(True)     # Run code
    .image_generation({
        "enabled": True,
        "app_ref": "infsh/flux@latest"
    })
    .build()
)

Internal Tool Options

Handling Tool Calls

Basic Handler

def handle_tool(call):
    if call.name == "greet":
        result = f"Hello, {call.args['name']}!"
    elif call.name == "calculate":
        result = eval(call.args['expression'])
    else:
        result = {"error": f"Unknown tool: {call.name}"}

    agent.submit_tool_result(call.id, result)

response = agent.send_message(
    "Greet John",
    on_tool_call=handle_tool
)

With Approval

def handle_tool(call):
    if call.requires_approval:
        print(f"Tool: {call.name}")
        print(f"Args: {call.args}")
        approved = input("Approve? (y/n): ").lower() == 'y'

        if not approved:
            agent.submit_tool_result(call.id, {
                "error": "Denied by user"
            })
            return

    result = execute_tool(call.name, call.args)
    agent.submit_tool_result(call.id, result)

Widget Results

Return structured data for UI widgets:

def handle_tool(call):
    if call.name == "confirm_order":
        # Return widget data
        agent.submit_tool_result(call.id, {
            "action": {"type": "confirm"},
            "form_data": {
                "order_id": "12345",
                "items": ["Widget A", "Widget B"],
                "total": 99.99
            }
        })

Complete Example

from inferencesh import (
    inference, tool, app_tool, webhook_tool,
    string, number, integer, boolean, enum_of,
    array, obj, optional, internal_tools
)

client = inference(api_key="inf_...")

# Calculator tool
calculator = (
    tool("calculate")
    .display("Calculator")
    .describe("Perform mathematical calculations")
    .param("expression", string("Math expression to evaluate"))
    .build()
)

# Image generation tool
image_gen = (
    app_tool("generate_image", "infsh/flux-schnell@latest")
    .describe("Generate an image from text")
    .param("prompt", string("Image description"))
    .param("style", enum_of(["realistic", "artistic", "cartoon"], "Image style"))
    .setup({"quality": "high"})
    .input({"steps": 20})
    .require_approval()
    .build()
)

# Slack notification tool
slack = (
    webhook_tool("notify", "https://hooks.slack.com/...")
    .describe("Send Slack notification")
    .param("message", string("Message to send"))
    .build()
)

# Built-in tools
internals = (
    internal_tools()
    .web_search(True)
    .code_execution(True)
    .build()
)

# Create agent with all tools
agent = client.agent({
    "core_app": {"ref": "infsh/claude-sonnet-4@latest"},
    "system_prompt": "You are a helpful assistant with various capabilities.",
    "tools": [calculator, image_gen, slack],
    "internal_tools": internals,
    "temperature": 0.7
})

# Handle tool calls
def handle_tool(call):
    if call.name == "calculate":
        try:
            result = eval(call.args["expression"])
            agent.submit_tool_result(call.id, {"result": result})
        except Exception as e:
            agent.submit_tool_result(call.id, {"error": str(e)})
    elif call.requires_approval:
        approved = input(f"Allow {call.name}? (y/n): ").lower() == 'y'
        if approved:
            # Let the app/webhook tool execute
            pass
        else:
            agent.submit_tool_result(call.id, {"error": "Denied"})

response = agent.send_message(
    "Calculate 15% tip on $85, then notify Slack",
    on_tool_call=handle_tool
)