uss-aigent 0.1.3

A progressive-disclosure Python library for LLM APIs

Aigent

A progressive-disclosure Python library for LLM APIs. Start with a one-liner, scale to full control — all in idiomatic Python.

from aigent import Aigent

agent = Aigent(api_type="openai")
with agent.session() as s:
    print(s.chat("Hello!"))

Installation

pip install uss-aigent

Requires Python 3.10+. The only external dependency is httpx.

Quick Start

Tier 1 — Chat

from aigent import Aigent

# Zero-config: reads OPENAI_API_KEY / ANTHROPIC_API_KEY from env
agent = Aigent(api_type="anthropic")

with agent.session(system="You are a professional translator.") as s:
    result = s.chat("Translate to English: 你好世界")
    print(result)

Tier 2 — Tools

from aigent import Aigent, tool

@tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """Get current weather for a city."""
    # In real code, call a weather API here
    return f"{city}: 22°{unit[0].upper()}"

agent = Aigent(api_type="openai")
with agent.session(tools=[get_weather]) as s:
    reply = s.chat("What's the weather in Beijing?")
    print(reply)

The @tool decorator auto-generates JSON Schema from your function signature and docstring — no manual schema writing.

Want to intercept tool calls? Use hooks:

# Before hook: inject user context into every tool call
def inject_user(args, tool_name):
    args.setdefault("user_id", current_user.id)
    return args

# After hook: result is the raw return type (dict, int, list, …), not a string!
def log_result(args, result, error, tool_name):
    if isinstance(result, dict) and result.get("success"):
        print(f"[OK] {tool_name}: {result}")
    return result

with agent.session(tools=[get_weather]) as s:
    s.hook_all("before", inject_user)   # runs before ALL tools
    s.hook(get_weather, "after", log_result)  # runs after this specific tool
    s.chat("What's the weather in Beijing?")

Tier 3 — Full Control

from aigent import Aigent, Message

agent = Aigent(api_type="anthropic")

# Manually orchestrate messages
resp = agent.raw(
    [Message.system("You are helpful."), Message.user("Hi!")],
    max_tokens=200,
    temperature=0.7,
)
print(resp.content)   # str
print(resp.usage)     # Usage(prompt_tokens=..., completion_tokens=...)

Session & Role API

agent = Aigent()

with agent.session() as s:
    # Chat as user (default)
    s.chat("Hello")

    # Chat as assistant
    s.chat("I'm doing great!", role="assistant")

    # Streaming
    for token in s.chat("Write a poem", stream=True):
        print(token, end="")

    # Insert without sending — build context gradually
    s.user.insert("I want to learn Python.")
    s.assistant.insert("Great choice! Where would you like to start?")
    reply = s.chat()  # sends existing history without adding new message

    # Role objects
    s.system.insert("You are a Python expert.")
    s.role("tool").insert('{"result": 42}', tool_call_id="call_abc")

    # Timeout — per-request override (chat > session > agent default)
    s.chat("quick question")                # uses agent/session default
    s.chat("complex task", timeout=180.0)   # 3 minutes for this call alone

    # Tool hooks — intercept before/after each tool call
    s.hook(my_tool, "before", validate_args)
    s.hook_all("after", log_all_results)

    # History management
    print(s.history)   # read-only list of messages
    s.clear()           # reset the conversation

Tool Hooks

Intercept tool calls with before and after hooks. Perfect for logging, parameter injection, result formatting, and error fallback.

from aigent import Aigent, tool

@tool
def search(query: str) -> dict:
    """Search a knowledge base."""
    return {"results": ["doc1", "doc2"], "count": 2}

@tool
def calculate(expr: str) -> float:
    """Evaluate a math expression."""
    return eval(expr)

agent = Aigent(api_type="openai")

# ── Per-tool hooks ──
with agent.session(tools=[search, calculate]) as s:
    s.hook(search, "before", lambda args, tn: {**args, "query": args["query"].strip()})
    s.hook(calculate, "after", lambda args, res, err, tn: round(res, 2) if not err else "error")
    s.chat("Find docs about Python and compute 3.14 * 2")

# ── Global hooks (apply to ALL tools) ──
def log_everything(args, result, error, tool_name):
    """result is the raw return type — dict, int, list, whatever the tool returns."""
    status = "FAIL" if error else "OK"
    print(f"[{status}] {tool_name}({args}) → {result}")
    return result

with agent.session(tools=[search, calculate]) as s:
    s.hook_all("after", log_everything)
    s.chat("Search for Python and compute 42 * 7")

Hook signatures:

• before(args: dict, tool_name: str) -> dict — modify/validate arguments
• after(args: dict, result: Any, error: Exception | None, tool_name: str) -> Any — process result, handle errors

Execution order: specific-tool hooks run before global hooks. before chain → tool execution → after chain.

Supported Backends

api_type	Backend	Default Model
"openai"	OpenAI Chat Completions	gpt-4o
"anthropic"	Anthropic Messages	claude-sonnet-4-6

OpenAI-compatible services (Azure, local LLMs, etc.) work via api_type="openai" with a custom base_url.

Configuration

agent = Aigent(
    api_type="openai",
    api_key="sk-...",                  # or OPENAI_API_KEY env var
    base_url="https://api.openai.com/v1",  # or OPENAI_BASE_URL env var
    model="gpt-4o",                   # or OPENAI_MODEL env var
    system="You are helpful.",         # default system prompt for all sessions
    timeout=30.0,                      # float (total seconds) or httpx.Timeout
    max_retries=3,
)

Timeout priority chain: chat(timeout=...) > session(timeout=...) > agent(timeout=...). Timeout also accepts httpx.Timeout objects for fine-grained control:

from httpx import Timeout

# Separate connect/read/write timeouts — great for tool-heavy workflows
agent = Aigent(
    api_type="openai",
    timeout=Timeout(connect=10.0, read=120.0, write=10.0),
)

# Per-session override
with agent.session(timeout=90.0, tools=[...]) as s:
    s.chat("normal task")               # uses session's 90s
    s.chat("heavy computation", timeout=300.0)  # 5 minutes for this one

Environment variables per backend:

Env	OpenAI	Anthropic
API key	OPENAI_API_KEY	ANTHROPIC_API_KEY
Base URL	OPENAI_BASE_URL	ANTHROPIC_BASE_URL
Model	OPENAI_MODEL	ANTHROPIC_MODEL

Error Handling

from aigent import (
    Aigent,
    AuthenticationError,
    RateLimitError,
    APIError,
    ConnectionError,
    TimeoutError,
)

agent = Aigent()
try:
    with agent.session() as s:
        s.chat("Hello")
except AuthenticationError:
    print("Check your API key.")
except RateLimitError:
    print("Slow down.")
except (ConnectionError, TimeoutError):
    print("Network issue.")
except APIError as e:
    print(f"API returned {e.status_code}")

All exceptions inherit from LLMError.

Streaming

with agent.session() as s:
    for token in s.chat("Tell me a story", stream=True):
        print(token, end="", flush=True)
    # Tokens are also accumulated and auto-appended to history

# Streaming with tools: TRUE streaming, no degradation.
# Text tokens appear in real time while tool_calls accumulate in the background.
with agent.session(tools=[get_weather]) as s:
    for token in s.chat("What's the weather in Beijing?", stream=True):
        print(token, end="", flush=True)
        # User sees "Let me check the weather..." immediately,
        # not after the entire tool-call round-trip.

License

Apache 2.0 — see LICENSE.