isA Agent SDK

pip install isa-agent-sdk

from isa_agent_sdk import query
 
async for msg in query("Hello, world!"):
    print(msg.content, end="" if msg.is_text else "\n")

# Streaming - get messages as they arrive
async for msg in query("Explain AI"):
    if msg.is_text:
        print(msg.content, end="")
    elif msg.is_tool_use:
        print(f"[Using: {msg.tool_name}]")
 
# Single mode - get complete response
from isa_agent_sdk import ask
 
result = await ask("What is 2+2?")
print(result.content)

from isa_agent_sdk import query, ISAAgentOptions
 
options = ISAAgentOptions(
    model="gpt-4o-mini",
    allowed_tools=["web_search", "read_file"],
    max_iterations=30,
    skills=["code-review", "debug"]
)
 
async for msg in query("Review this code", options=options):
    print(msg.content, end="" if msg.is_text else "\n")

from isa_agent_sdk import execute_tool, get_available_tools
 
# Discover tools
tools = await get_available_tools(user_query="search the web")
 
# Execute directly
result = await execute_tool("web_search", {"query": "Python tutorials"})

from isa_agent_sdk import request_tool_permission
 
authorized = await request_tool_permission(
    tool_name="delete_file",
    tool_args={"path": "/data/file.txt"},
    reason="Cleaning up old data"
)
 
if authorized:
    # proceed
    pass

options = ISAAgentOptions(
    skills=["code-review", "debug", "refactor"]
)

from isa_agent_sdk import register_trigger, TriggerType
 
trigger_id = await register_trigger(
    user_id="user-123",
    trigger_type=TriggerType.THRESHOLD,
    description="Alert on price drop",
    conditions={"threshold_value": 5.0, "direction": "down"},
    action_config={"prompt": "Analyze the drop"}
)

# Via agent conversation
"Create a todo task called 'Review PR' with high priority"
"What's on my calendar tomorrow?"
"Alert me when Bitcoin drops 5%"
 
# Programmatic access
options = ISAAgentOptions(
    user_id="user-123",
    allowed_tools=[
        "create_task", "list_tasks", "complete_task",
        "create_calendar_event", "get_upcoming_events",
        "register_price_alert", "list_triggers"
    ]
)

from pydantic import BaseModel
from isa_agent_sdk import query, ISAAgentOptions, OutputFormat
 
class Recipe(BaseModel):
    name: str
    ingredients: list[str]
    prep_time_minutes: int
 
async for msg in query(
    "Find a chocolate chip cookie recipe",
    options=ISAAgentOptions(
        output_format=OutputFormat.from_pydantic(Recipe)
    )
):
    if msg.has_structured_output:
        recipe = msg.parse(Recipe)
        print(f"Recipe: {recipe.name}")

from isa_agent_sdk import query, ISAAgentOptions, ExecutionEnv
 
options = ISAAgentOptions(
    env=ExecutionEnv.DESKTOP,
    user_id="xenodennis",
    allowed_tools=["read_file", "write_file", "bash_execute", "glob_files"]
)
 
async for msg in query("Find all Python files and search for 'API'", options=options):
    if msg.is_tool_use:
        print(f"[Executing on desktop: {msg.tool_name}]")
    elif msg.is_text:
        print(msg.content, end="")

SDK → Pool Manager → Desktop Agent → Local Filesystem

from isa_agent_sdk import query, resume, ExecutionMode
 
options = ISAAgentOptions(
    execution_mode=ExecutionMode.COLLABORATIVE,
    session_id="my-task-123"
)
 
# Later, resume from checkpoint
async for msg in resume("my-task-123"):
    print(msg.content)

from isa_agent_sdk.services.voice import VoiceOrchestrator, VoiceSessionConfig
 
orchestrator = VoiceOrchestrator(
    realtime_gateway_url="ws://localhost:8082/v1/realtime",
    config=VoiceSessionConfig(
        audio_format="pcm16",
        sample_rate=16000,
        barge_in_enabled=True,
        turn_detection_threshold=0.5,
    ),
)
 
await orchestrator.start_session()
await orchestrator.process_audio(audio_chunk)
await orchestrator.handle_barge_in()
await orchestrator.stop_session()

from isa_agent_sdk import Agent, ISAAgentOptions
from isa_agent_sdk.agents.swarm import SwarmOrchestrator
 
researcher = Agent(name="researcher", options=ISAAgentOptions(
    allowed_tools=["web_search"], system_prompt="You research topics."
))
writer = Agent(name="writer", options=ISAAgentOptions(
    system_prompt="You write articles based on research."
))
 
swarm = SwarmOrchestrator(
    agents={"researcher": researcher, "writer": writer},
    entry_agent="researcher",
    max_handoffs=10,
)
 
# Sequential handoffs
result = await swarm.run("Write an article about quantum computing")
print(result.text)
print(f"Handoff trace: {result.handoff_trace}")
 
# DAG execution (parallel where possible)
result = await swarm.run_dag(tasks=[...], parallel=True)

from isa_agent_sdk.delegation import (
    DelegationClient, TeamDefinition, TeamRegistry, DelegationRequest
)
 
registry = TeamRegistry()
registry.register(TeamDefinition(
    name="data-team",
    description="Handles data analysis tasks",
    skills=["data-analysis", "visualization"],
    url="http://data-agents:8102",
    auth_token_env="DATA_TEAM_TOKEN",
))
 
client = DelegationClient(registry, circuit_breaker_threshold=3)
result = await client.delegate(DelegationRequest(
    team_name="data-team",
    task="Analyze Q4 sales trends",
    context={"dataset": "sales_2025"},
))
print(f"Status: {result.status}, Duration: {result.duration_ms}ms")

from isa_agent_sdk.agents.lifecycle import AgentLifecycle
 
lifecycle = AgentLifecycle(
    name="my-worker",
    drain_timeout=30.0,
    heartbeat_interval=30.0,
    on_start=lambda: print("Started"),
    on_stop=lambda: print("Stopped"),
)
 
await lifecycle.start()
 
# Track in-flight work — stop() waits for tracked work to complete
async with lifecycle.track_work():
    await process_task()
 
print(lifecycle.health())  # {"state": "RUNNING", "uptime_seconds": ..., "in_flight_count": 0}
 
await lifecycle.stop()  # Graceful drain, handles SIGTERM/SIGINT

from isa_agent_sdk.agents.review import writer_reviewer_run
 
result = await writer_reviewer_run(
    writer=writer_agent,
    reviewer=reviewer_agent,
    prompt="Write a technical blog post about microservices",
    max_rounds=2,
    review_instructions="Check for technical accuracy and clarity",
)
 
print(f"Approved: {result.approved}")
print(f"Rounds: {result.rounds_completed}")
print(result.final_text)

from isa_agent_sdk.services.auto_detection.loop_detector import LoopDetector
 
detector = LoopDetector(
    repetition_threshold=3,
    cycle_min_occurrences=2,
    convergence_threshold=0.9,
)
 
detection = detector.check_tool_loop(tool_calls)
if detection:
    print(f"Loop type: {detection.loop_type}")  # tool_repetition, tool_cycle
    print(f"Confidence: {detection.confidence}")
    print(f"Suggestion: {detection.suggestion}")

from isa_agent_sdk.services.observability.metrics import (
    record_node_execution,
    record_model_call,
    record_tool_execution,
    metrics_text,
)
 
# Automatically instrumented when prometheus_client is installed
# Metrics include:
# - agent_node_duration_seconds, agent_node_executions_total
# - agent_model_call_duration_seconds, agent_model_tokens_total
# - agent_tool_duration_seconds, agent_tool_executions_total
# - agent_pool_acquire_duration_seconds, agent_pool_active_vms
# - agent_hil_interrupts_total, agent_checkpoint_operations_total
 
# Expose metrics endpoint
metrics_bytes = metrics_text()

from isa_agent_sdk.core.resilience.tls import create_tls_client, get_uvicorn_tls_kwargs
 
# Client with mTLS
client = create_tls_client(
    ca_bundle="/certs/ca.pem",
    client_cert="/certs/client.pem",
    client_key="/certs/client-key.pem",
)
 
# Server with TLS
uvicorn_kwargs = get_uvicorn_tls_kwargs(
    server_cert="/certs/server.pem",
    server_key="/certs/server-key.pem",
)

from isa_agent_sdk.core.resilience.rate_limiter import RateLimitMiddleware, UserTier
 
# Tier limits (requests per minute / daily):
# FREE: 10/min, 100/day
# STARTER: 30/min, 1000/day
# PRO: 60/min, 5000/day
# ENTERPRISE: 120/min, 50000/day
 
# Add as FastAPI middleware — tier resolved from X-User-Tier header or JWT claim
app.add_middleware(RateLimitMiddleware)

from isa_agent_sdk.services.proxy import SubscriptionProxyManager
 
manager = SubscriptionProxyManager()
await manager.register_provider("anthropic-sub", config={...})
await manager.start_provider("anthropic-sub")
 
health = await manager.check_health()  # HEALTHY, DEGRADED, or UNHEALTHY

# 1. Create deployment
curl -X POST http://localhost:8095/api/v1/deployments \
  -H "X-User-ID: user123" \
  -d '{"app_name": "my-agent"}'
 
# 2. Upload code (auto-builds)
tar -czf my-agent.tar.gz -C my-agent .
curl -X POST http://localhost:8095/api/v1/deployments/{id}/upload \
  -F "file=@my-agent.tar.gz"
 
# 3. Deploy version
curl -X POST http://localhost:8095/api/v1/deployments/{id}/deploy \
  -d '{"version": "v1.0.0", "replicas": 2}'
 
# 4. Query your deployed agent
curl -X POST http://localhost:8095/api/v1/apps/{id}/query \
  -d '{"prompt": "Hello from the cloud!"}'

from isa_agent_sdk import ISAAgentClient
 
async with ISAAgentClient() as client:
    await client.query("Remember the number 42")
    async for msg in client.receive():
        print(msg.content)
 
    # Continue conversation
    await client.query("What number did I mention?")
    async for msg in client.receive():
        print(msg.content)  # Will recall 42
 
    # Or use convenience method
    response = await client.ask("What is 2 + 2?")

from isa_agent_sdk import ISAAgent
 
client = ISAAgent(base_url="http://localhost:8000")
 
# Simple chat
response = client.chat.create(message="Hello!", user_id="user123")
 
# Streaming
for event in client.chat.stream(message="Explain AI"):
    if event.is_content:
        print(event.content, end="")

from isa_agent_sdk import query, ISAAgentOptions
 
options = ISAAgentOptions(
    skills=["code-review"],
    allowed_tools=["read_file"]
)
 
async for msg in query("Review main.py for issues", options=options):
    print(msg.content, end="" if msg.is_text else "\n")

options = ISAAgentOptions(
    allowed_tools=["web_search", "fetch_url"],
    max_iterations=50
)
 
async for msg in query("Research latest AI developments", options=options):
    if msg.is_tool_use:
        print(f"[Searching: {msg.tool_args}]")
    elif msg.is_text:
        print(msg.content, end="")

options = ISAAgentOptions(
    execution_mode=ExecutionMode.COLLABORATIVE,
    allowed_tools=["read_file", "write_file", "bash"]
)
 
async for msg in query("Process all CSV files in /data", options=options):
    if msg.is_checkpoint:
        await msg.respond({"continue": True})
    elif msg.is_text:
        print(msg.content, end="")

from isa_agent_sdk import (
    ISASDKError,        # Base error
    ConnectionError,    # Service connection failed
    ToolExecutionError, # Tool failed
    SessionError,       # Session issues
    ValidationError,    # Input validation
)
 
try:
    async for msg in query("Do something"):
        ...
except ToolExecutionError as e:
    print(f"Tool {e.tool_name} failed: {e.message}")
except SessionError as e:
    print(f"Session {e.session_id} error: {e}")
except ISASDKError as e:
    print(f"SDK error: {e}")