argentor-sdk v1.4.7

Python SDK client for the Argentor AI agent framework REST API.

Installation

pip install argentor-sdk

For development:

pip install argentor-sdk[dev]

Quick Start

from argentor import ArgentorClient

client = ArgentorClient(
    base_url="http://localhost:8080",
    api_key="your-api-key",
)

# Run a task
result = client.run_task(
    role="code_reviewer",
    context="Review the following pull request...",
)
print(result)

# Stream results via SSE
for chunk in client.run_task_stream(
    role="assistant",
    context="Explain how Argentor works",
):
    print(chunk, end="", flush=True)

# Batch execution
batch = client.batch_tasks(
    tasks=[
        {"agent_role": "analyst", "context": "Analyze Q1 sales data"},
        {"agent_role": "analyst", "context": "Analyze Q2 sales data"},
    ],
    max_concurrent=5,
)
print(batch)

# Evaluate a response
evaluation = client.evaluate(
    text="The code looks clean and follows best practices.",
    context="Code review task",
    criteria=["accuracy", "completeness", "helpfulness"],
)
print(evaluation)

# List skills
skills = client.list_skills()
for skill in skills:
    print(skill["name"])

# Execute a skill
result = client.execute_skill("echo", {"text": "Hello, world!"})
print(result)

# Health check
health = client.health()
print(health)

Async Usage

import asyncio
from argentor import AsyncArgentorClient

async def main():
    async with AsyncArgentorClient(
        base_url="http://localhost:8080",
        api_key="your-api-key",
    ) as client:
        # Run a task
        result = await client.run_task(
            role="assistant",
            context="Hello, world!",
        )
        print(result)

        # Stream results
        async for chunk in client.run_task_stream(
            role="assistant",
            context="Explain Argentor",
        ):
            print(chunk)

        # List sessions
        sessions = await client.list_sessions()
        print(sessions)

asyncio.run(main())

Agent SDK (subprocess wrapper)

The agent module wraps the argentor CLI binary as a subprocess, similar to how Claude Agent SDK wraps Claude Code. It communicates via NDJSON over stdin/stdout and works with any LLM provider.

import asyncio
from argentor import query, query_simple, AgentOptions

# Stream events from the agent
async def main():
    async for event in query(
        "What files are in this directory?",
        AgentOptions.claude("sk-your-api-key"),
    ):
        if event.type == "assistant":
            print(event.text)
        elif event.type == "tool_use":
            print(f"[tool] {event.data.get('name')}")
        elif event.is_done:
            print(f"\nDone: {event.text}")

asyncio.run(main())

One-liner convenience functions

from argentor import ask_claude, ask_openai, ask_gemini, ask_ollama

# Each returns the final text output
result = asyncio.run(ask_claude("Explain Argentor", "sk-..."))
result = asyncio.run(ask_openai("Explain Argentor", "sk-..."))
result = asyncio.run(ask_gemini("Explain Argentor", "AIza..."))
result = asyncio.run(ask_ollama("Explain Argentor", "llama3"))

Provider presets

AgentOptions.claude(api_key)   # Anthropic Claude
AgentOptions.openai(api_key)   # OpenAI GPT-4o
AgentOptions.gemini(api_key)   # Google Gemini
AgentOptions.ollama("llama3")  # Local Ollama (no key)

AgentOptions

Error Handling

from argentor import ArgentorClient
from argentor.exceptions import ArgentorAPIError, ArgentorConnectionError

client = ArgentorClient(base_url="http://localhost:8080")

try:
    result = client.run_task(role="assistant", context="Hello")
except ArgentorAPIError as e:
    print(f"API error {e.status_code}: {e.message}")
    print(f"Response body: {e.response_body}")
except ArgentorConnectionError as e:
    print(f"Connection failed: {e.message}")

API Reference

ArgentorClient / AsyncArgentorClient

Both clients expose the same methods. The async client returns awaitables and uses async for for streaming.

License

AGPL-3.0-only