Quickstart

import os
from polos import (
    Agent, max_steps, MaxStepsConfig,
    sandbox_tools, SandboxToolsConfig, LocalEnvironmentConfig,
)

workspace_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "workspace")

# Create sandbox tools that run locally on the host
tools = sandbox_tools(
    SandboxToolsConfig(
        env="local",
        local=LocalEnvironmentConfig(
            cwd=workspace_dir,
            path_restriction=workspace_dir,  # prevent file access outside workspace
        ),
    )
)

# Define an agent that can write and run code locally
coding_agent = Agent(
    id="local_coding_agent",
    provider="anthropic",
    model="claude-sonnet-4-5",
    system_prompt=(
        f"You are a coding agent with access to the local filesystem. "
        f"You can create files, edit code, run shell commands, and search the codebase. "
        f"Your workspace is at {workspace_dir}. "
        f"Use the tools to complete the task, then summarize what you did and show the output. "
        f"Always verify your work by running the code after writing it. "
        f"In your final response, include the actual output from running the code."
    ),
    tools=tools,
    stop_conditions=[max_steps(MaxStepsConfig(count=30))],
)

import asyncio
import os
from dotenv import load_dotenv
from polos import PolosClient, Worker
from agents import coding_agent, tools

load_dotenv()

async def main():
    client = PolosClient(
        project_id=os.getenv("POLOS_PROJECT_ID"),
        api_url=os.getenv("POLOS_API_URL", "http://localhost:8080"),
    )

    worker = Worker(
        client=client,
        agents=[coding_agent],
        tools=list(tools),
    )

    print("Starting worker...")
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

from polos import PolosClient
from polos.features import events

# Invoke the agent
handle = await client.invoke(coding_agent.id, {
    "input": "Create hello.js that prints 'Hello, world!' and run it.",
    "streaming": True,
})

# Stream events - approve each exec/write/edit when the agent suspends
async for event in events.stream_workflow(client, handle.root_workflow_id, handle.id):
    if event.event_type and event.event_type.startswith("suspend_"):
        step_key = event.event_type[len("suspend_"):]
        form = event.data.get("_form", {})
        context = form.get("context", {})
        print(f"\n  Agent wants to: {context.get('command') or context.get('tool', step_key)}")
        approved = input("  Approve? (y/n): ").strip().lower() == "y"
        await client.resume(handle.root_workflow_id, handle.id, step_key, {"approved": approved})