Metadata-Version: 2.4
Name: a3s-code
Version: 1.7.0
Classifier: Programming Language :: Rust
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Summary: A3S Code - Native Python bindings for the AI coding agent
License: MIT
Requires-Python: >=3.8
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM

# A3S Code — Python SDK

Native Python bindings for the A3S Code AI coding agent, built with PyO3.

## Installation

```bash
pip install a3s-code
```

## Quick Start

```python
from a3s_code import Agent

agent = Agent.create("agent.hcl")
session = agent.session("/my-project")

result = session.send("What files handle authentication?")
print(result.text)
```

## Slash Commands

Every session includes built-in slash commands dispatched before the LLM:

```python
# List all available commands
commands = session.list_commands()
for cmd in commands:
    print(f"/{cmd['name']:15s} {cmd['description']}")

# Built-in commands
result = session.send("/help")       # List all commands
result = session.send("/model")      # Show current model
result = session.send("/cost")       # Token usage and cost
result = session.send("/history")    # Conversation stats
result = session.send("/cron-list")  # List scheduled tasks
```

### Custom Commands

```python
def my_handler(args: str, ctx: dict) -> str:
    return f"Model: {ctx['model']}, History: {ctx['history_len']} msgs, args: {args!r}"

session.register_command("status", "Show session info", my_handler)
result = session.send("/status hello")
```

## Scheduled Tasks

Schedule recurring prompts that fire after each `send()` call:

```python
# Via /loop slash command
r = session.send("/loop 30s check deployment status")
print(r.text)  # Scheduled [a1b2c3d4]: "check deployment status" — fires every 30s

# Programmatic API
task_id = session.schedule_task("summarize recent commits", 300)  # every 5 min

# List active tasks
for t in session.list_scheduled_tasks():
    print(f"[{t['id']}] every {t['interval_secs']}s — \"{t['prompt']}\"")

# Cancel
session.cancel_scheduled_task(task_id)
session.send(f"/cron-cancel {task_id}")
```

**Interval syntax:** `30s`, `5m`, `2h`, `1d`. Leading or trailing with `every` clause.

## Full API

```python
from a3s_code import (
    Agent,
    SessionOptions,
    DefaultSecurityProvider,
    DocumentOcrProvider,
    FileMemoryStore,
    FileSessionStore,
)

agent = Agent.create("agent.hcl")
session = agent.session("/my-project",
    model="openai/gpt-4o",
    builtin_skills=True,
    planning=True,
)

# Send / Stream
result = session.send("Explain the auth module")
for event in session.stream("Refactor auth"):
    if event.event_type == "text_delta":
        print(event.text, end="", flush=True)

# Direct tools (bypass LLM)
session.read_file("src/main.py")
session.bash("pytest")
session.glob("**/*.py")
session.grep("TODO")

# Rich document parsing metadata
tool = session.tool("agentic_parse", {"path": "docs/scanned.pdf"})
print(tool.metadata)  # parsed dict
print(tool.document_runtime)  # parsed dict
runtime = tool.document_runtime_info
if runtime and runtime.ocr:
    print(runtime.ocr.provider, runtime.ocr.model, runtime.ocr.dpi)

query_tool = session.tool(
    "agentic_parse",
    {"path": "docs/scanned.pdf", "query": "overview"},
)
for block in query_tool.agentic_parse_llm_blocks_info:
    location = block.location.display if block.location else None
    print(block.index, block.kind, block.label, location)

search = session.tool("agentic_search", {"query": "invoice total", "mode": "fast"})
for result in search.agentic_search_results_info:
    if result.document_runtime and result.document_runtime.ocr:
        print(result.path, result.document_runtime.ocr.provider)
    for match in result.matches:
        print(match.line_number, match.locator, match.content)

# Slash commands & scheduling
session.list_commands()
session.register_command("ping", "Pong!", lambda args, ctx: "pong")
task_id = session.schedule_task("daily report", 86400)
session.list_scheduled_tasks()
session.cancel_scheduled_task(task_id)

# Memory
session.remember_success("task", ["tool"], "result")
session.recall_similar("auth", 5)

# Hooks
session.register_hook("audit", "pre_tool_use", handler_fn)

# MCP
session.add_mcp_server("github", command="npx", args=["-y", "@modelcontextprotocol/server-github"])
session.mcp_status()
session.tool_names()
session.remove_mcp_server("github")

# Persistence
opts = SessionOptions()
opts.session_store = FileSessionStore('./sessions')
opts.session_id = 'my-session'
opts.auto_save = True
session2 = agent.session(".", opts)
resumed = agent.resume_session('my-session', opts)
```

## OCR Backend For Better Context Extraction

Use `DocumentOcrProvider` when you want `agentic_search` and `agentic_parse`
to extract context from scanned PDFs or images via a Python OCR callback.

```python
from a3s_code import Agent, DocumentParserConfig, DocumentOcrConfig
from a3s_code import DocumentOcrProvider, DocumentParserRegistry, SessionOptions


def ocr_callback(request: dict) -> str | None:
    print("OCR request:", request["path"], request["format"])
    print("OCR config:", request["config"])
    return "Scanned invoice total: 42 USD"


agent = Agent.create("agent.hcl")
opts = SessionOptions()
opts.document_parser_registry = DocumentParserRegistry(
    DocumentParserConfig()
)
opts.document_ocr_provider = DocumentOcrProvider(
    "python-mock-ocr",
    ocr_callback,
    formats=["pdf", "image"],
    model="openai/gpt-4.1-mini",
)

session = agent.session(".", opts)
tool = session.tool("agentic_parse", {"path": "docs/scanned.pdf"})
runtime = tool.document_runtime_info

if runtime and runtime.ocr:
    print(runtime.ocr.used, runtime.ocr.provider, runtime.ocr.format)
```

The callback receives:

- `path`: absolute file path selected by the document parser
- `format`: `pdf`, `docx`, `xlsx`, `pptx`, `odf`, or `image`
- `config`: normalized OCR config with `enabled`, `model`, `prompt`, `max_images`, `dpi`

Return a string to provide OCR text, or `None` to signal no OCR result.

## Agentic Search Match Locators

`agentic_search_results_info` now exposes typed match and sampled-line entries
so you can inspect page / section locators without parsing raw metadata.

```python
search = session.tool("agentic_search", {"query": "overview", "mode": "fast"})

for result in search.agentic_search_results_info:
    for match in result.matches:
        print(match.line_number, match.locator, match.content)
        if match.context_before:
            print("before:", match.context_before[-1])

deep = session.tool("agentic_search", {"query": "overview", "mode": "deep"})
for result in deep.agentic_search_results_info:
    for sampled in result.sampled_lines:
        print(sampled.line_number, sampled.locator, sampled.distance, sampled.weight)
```

## Agentic Parse LLM Blocks

When `agentic_parse` runs with a `query`, the SDK also exposes which structured
document blocks were actually selected for the LLM input.

```python
tool = session.tool(
    "agentic_parse",
    {"path": "docs/scanned.pdf", "query": "overview"},
)

for block in tool.agentic_parse_llm_blocks_info:
    location = block.location.display if block.location else None
    print(block.index, block.kind, block.label, location)
```

## Sub-Agent Events

`Orchestrator.create(agent=agent)` creates real LLM-backed sub-agents. Use
`handle.events()` to observe sub-agent progress, tool calls, and streamed text.

```python
from a3s_code import Agent, Orchestrator, SubAgentConfig

agent = Agent.create("agent.hcl")
orch = Orchestrator.create(agent=agent)
handle = orch.spawn_subagent(SubAgentConfig(
    agent_type="general",
    prompt="Use bash to print hello, then explain it.",
    permissive=True,
    max_steps=5,
))

events = handle.events()
while True:
    event = events.recv(timeout_ms=1000)
    if event is None:
        continue

    event_type = event["event_type"]

    if event_type == "sub_agent_internal_event" and event.get("type") == "text_delta":
        print(event.get("text", ""), end="", flush=True)
    elif event_type == "tool_execution_started":
        print("tool args:", event["args"])
    elif event_type == "tool_execution_completed":
        print("tool duration_ms:", event["duration_ms"])
    elif event_type == "sub_agent_completed":
        break
```

Important details:

- Event names use `sub_agent_*`, not `subagent_*`.
- `sub_agent_internal_event` payloads are flattened. For example, a text delta is:
  `{"event_type": "sub_agent_internal_event", "type": "text_delta", "text": "..."}`
  rather than nesting under `event`.
- `tool_execution_started.args` contains the accumulated tool input.
- `tool_execution_completed.duration_ms` is guaranteed to be at least `1` for real tool calls.

## License

MIT

