Metadata-Version: 2.4
Name: infinium-o2
Version: 1.1.0
Summary: Professional Python SDK for Infinium API
Project-URL: Homepage, https://github.com/xyendigital/infinium-python-sdk
Project-URL: Documentation, https://github.com/xyendigital/infinium-python-sdk#readme
Project-URL: Repository, https://github.com/xyendigital/infinium-python-sdk
Project-URL: Bug Tracker, https://github.com/xyendigital/infinium-python-sdk/issues
Project-URL: Changelog, https://github.com/xyendigital/infinium-python-sdk/blob/main/CHANGELOG.md
Author-email: Infinium <support@i42m.ai>
Maintainer-email: Infinium <support@i42m.ai>
License: MIT
License-File: LICENSE
Keywords: agent,api,automation,client,infinium,sdk
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Internet :: WWW/HTTP :: HTTP Servers
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Networking
Requires-Python: >=3.9
Requires-Dist: httpx>=0.25.0
Requires-Dist: typing-extensions>=4.8.0; python_version < '3.12'
Provides-Extra: dev
Requires-Dist: black>=23.7.0; extra == 'dev'
Requires-Dist: flake8>=6.0.0; extra == 'dev'
Requires-Dist: isort>=5.12.0; extra == 'dev'
Requires-Dist: mypy>=1.5.0; extra == 'dev'
Requires-Dist: pre-commit>=3.3.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21.0; extra == 'dev'
Requires-Dist: pytest-cov>=4.1.0; extra == 'dev'
Requires-Dist: pytest-mock>=3.11.0; extra == 'dev'
Requires-Dist: pytest>=7.4.0; extra == 'dev'
Provides-Extra: examples
Requires-Dist: python-dotenv>=1.0.0; extra == 'examples'
Description-Content-Type: text/markdown

# Infinium Python SDK

The official Python SDK for the [Infinium](https://i42m.ai) agent observability platform. Send trace data from your AI agents to Infinium, where **Maestro** (our behavioral intelligence engine) analyzes and scores agent performance.

Your agent does work -- calls LLMs, hits APIs, processes data. The SDK captures **what happened** (steps taken, tokens used, errors hit, time elapsed) and ships it to Infinium. Maestro then grades the work. The agent never judges itself; it just reports facts.

## Installation

```bash
pip install infinium-o2
```

**Requirements:** Python 3.9+

## Quick Start

```python
from infinium import InfiniumClient

client = InfiniumClient(
    agent_id="your-agent-id",
    agent_secret="your-agent-secret",
)
```

Send your first trace:

```python
client.send_task(
    name="Classify support ticket",
    description="Categorized an inbound support ticket by type and urgency.",
    duration=2.4,
    input_summary="Customer ticket about billing issue",
    output_summary="Category: billing, Urgency: high",
    llm_usage={"model": "gpt-4o", "provider": "openai", "prompt_tokens": 320, "completion_tokens": 45},
)
```

## Auto-Instrumentation

The fastest way to get full observability. `watch()` patches your LLM client to automatically capture every call, and `@client.trace()` wraps your function to bundle everything into a trace:

```python
from openai import OpenAI
from infinium import InfiniumClient
from infinium.integrations import watch

client = InfiniumClient(agent_id="...", agent_secret="...")
openai = watch(OpenAI())  # Patches this instance to auto-capture LLM calls

@client.trace("Summarize Article")
def summarize(article: str) -> str:
    resp = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Summarize this article in 3 bullet points."},
            {"role": "user", "content": article},
        ],
    )
    return resp.choices[0].message.content

# LLM tokens, latency, and timing are captured automatically -- trace is auto-sent
result = summarize("The Federal Reserve announced today...")
```

### Supported Providers

| Provider | Client | Detection |
|----------|--------|-----------|
| OpenAI | `openai.OpenAI` / `openai.AsyncOpenAI` | Module name |
| Anthropic | `anthropic.Anthropic` / `anthropic.AsyncAnthropic` | Module name |
| Google Gemini | `google.generativeai.GenerativeModel` | Class name |
| xAI (Grok) | `openai.OpenAI(base_url="https://api.x.ai/...")` | Base URL |

All providers support streaming transparently -- chunk accumulation and token extraction happen automatically.

### Content Capture

By default, `watch()` only captures metadata (model, tokens, latency). To also capture input/output previews (truncated to 500 chars):

```python
openai = watch(OpenAI(), capture_content=True)
```

## Sending Traces

There are three ways to send traces, from simplest to most detailed:

### 1. `send_task()` -- Simple, Inline

Best for single-step agents. Call your LLM, then report what happened:

```python
import time
from openai import OpenAI

openai = OpenAI()
start = time.time()
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": ticket_text}],
)
duration = time.time() - start
usage = response.usage

client.send_task(
    name="Classify support ticket",
    description="Categorized an inbound support ticket.",
    duration=duration,
    llm_usage={
        "model": "gpt-4o", "provider": "openai",
        "prompt_tokens": usage.prompt_tokens,
        "completion_tokens": usage.completion_tokens,
    },
    customer={"customer_name": customer_name},
)
```

### 2. `TaskData` -- Structured, Multi-Step

When your agent runs multiple steps, build a `TaskData` with execution details. This gives Maestro the most signal:

```python
from infinium import InfiniumClient, ExecutionStep, LlmCall, LlmUsage, ExpectedOutcome

steps = []

# Step 1: Search
steps.append(ExecutionStep(
    step_number=1, action="tool_use",
    description="Search knowledge base",
    duration_ms=340,
    output_preview=f"Found {len(results)} results",
))

# Step 2: Analyze with LLM
steps.append(ExecutionStep(
    step_number=2, action="llm_inference",
    description="Analyze search results",
    duration_ms=1200,
    llm_call=LlmCall(model="gpt-4o", provider="openai", prompt_tokens=800, completion_tokens=200),
))

td = InfiniumClient.create_task_data(
    name="Research: quarterly earnings",
    description="Multi-step research with source gathering and analysis",
    duration=3.2,
    steps=steps,
    expected_outcome=ExpectedOutcome(
        task_objective="Produce sourced analysis of quarterly earnings",
        required_deliverables=["Source list", "Analysis"],
    ),
    llm_usage=LlmUsage(model="gpt-4o", provider="openai", prompt_tokens=800, completion_tokens=200, api_calls_count=1),
)
client.send_task_data(td)
```

### 3. `TraceBuilder` -- Recommended for Real Agents

Steps are context managers that auto-capture timing and exceptions:

```python
from infinium import TraceBuilder, LlmUsage

trace = TraceBuilder("Blog Post Generator", "Generate a publish-ready blog post.")
trace.set_expected_outcome(
    task_objective="Produce a 600-word blog post with SEO metadata",
    required_deliverables=["Blog post", "SEO title", "Meta description"],
)
trace.set_input_summary(f"Topic: {topic}")

with trace.step("llm_inference", "Create outline") as step:
    resp = openai.chat.completions.create(model="gpt-4o", messages=[...])
    outline = resp.choices[0].message.content
    step.set_output(outline[:500])
    step.record_llm_call("gpt-4o", provider="openai",
                         prompt_tokens=resp.usage.prompt_tokens,
                         completion_tokens=resp.usage.completion_tokens)

with trace.step("llm_inference", "Write full post") as step:
    resp = openai.chat.completions.create(model="gpt-4o", messages=[...])
    blog_post = resp.choices[0].message.content
    step.set_output(blog_post[:500])
    step.record_llm_call("gpt-4o", provider="openai",
                         prompt_tokens=resp.usage.prompt_tokens,
                         completion_tokens=resp.usage.completion_tokens)

trace.set_output_summary(f"Generated {len(blog_post.split())}-word blog post.")
task_data = trace.build()  # Auto-computes total duration
client.send_task_data(task_data)
```

## `@trace_agent` Decorator

For single-function agents, the decorator handles everything -- duration, input/output capture, error recording, and sending:

```python
from infinium import trace_agent

@trace_agent("Email Classifier", client)
def classify_email(email_body: str) -> dict:
    resp = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Classify this email. Return JSON with intent, priority, department."},
            {"role": "user", "content": email_body},
        ],
        response_format={"type": "json_object"},
    )
    return json.loads(resp.choices[0].message.content)

# Every call auto-sends a trace
result = classify_email("Hi, I was charged twice for my subscription...")
```

If the function raises, the error is captured in the trace, the trace is still sent, and the exception re-raised. Use `@async_trace_agent` for async functions, or `@client.trace()` which auto-detects both.

## Prompt Studio

Retrieve managed prompts with optional variable substitution:

```python
prompt = client.get_prompt(
    prompt_id="your-prompt-id",
    prompt_key="your-prompt-key",
    version="latest",
    variables={"customer_name": "Acme Corp", "tone": "professional"},
)

print(prompt.content)           # Raw template
print(prompt.rendered_content)  # With variables substituted
```

When used inside a `@trace_agent` or `@client.trace()` decorated function, prompt fetches are automatically recorded in the trace.

## Async Support

Every feature has a sync and async variant:

```python
import asyncio
from infinium import AsyncInfiniumClient

async def main():
    async with AsyncInfiniumClient(agent_id="...", agent_secret="...") as client:
        await client.send_task(
            name="Moderate content",
            description="Check user content for policy violations.",
            duration=1.8,
        )

asyncio.run(main())
```

## Batch Sending

Send multiple traces at once:

```python
tasks = [InfiniumClient.create_task_data(...) for doc in documents]
result = client.send_tasks_batch(tasks)
print(f"Sent {result.successful}/{result.successful + result.failed} traces")
```

## Polling for Maestro Results

After sending a trace, poll until Maestro finishes analysis:

```python
response = client.send_task_data(task_data)
trace_id = response.data.get("traceId")

interpretation = client.wait_for_interpretation(trace_id, timeout=120, poll_interval=3)
print(interpretation.data["interpretedTraceResult"])
```

## Configuration

```python
client = InfiniumClient(
    agent_id="your-agent-id",
    agent_secret="your-agent-secret",
    base_url="https://platform.i42m.ai/api/v1",  # API endpoint (default)
    timeout=30.0,                                  # Request timeout in seconds
    max_retries=3,                                 # Retry attempts (exponential backoff)
    enable_rate_limiting=True,                     # Client-side rate limiting
    requests_per_second=10.0,                      # Rate limit threshold
    verify_ssl=True,                               # Set False for self-signed certs
)
```

## Error Handling

```python
from infinium.exceptions import (
    AuthenticationError, RateLimitError, ValidationError,
    NetworkError, ServerError, InfiniumTimeoutError,
)

try:
    client.send_task_data(task_data)
except AuthenticationError:
    print("Bad credentials -- check agent_id and agent_secret")
except RateLimitError as e:
    print(f"Rate limited -- retry after {e.retry_after}s")
except ValidationError as e:
    print(f"Invalid input: {e.field} -- {e}")
except NetworkError:
    print("Cannot reach the API")
except InfiniumTimeoutError:
    print("Request timed out")
```

The SDK automatically retries on 5xx errors, network failures, and timeouts (up to `max_retries`). Auth, validation, and rate limit errors are not retried.

## Data Types

### Trace Enrichment

| Type | Purpose |
|------|---------|
| `ExecutionStep` | One thing the agent did (tool call, LLM call, decision) |
| `ErrorDetail` | A factual error record (type, message, recoverable?) |
| `LlmCall` | A single LLM invocation (model, tokens, latency) |
| `ToolCall` | A tool/API invocation (name, status code, duration) |
| `ExpectedOutcome` | What the agent was asked to produce (Maestro's grading rubric) |
| `LlmUsage` | Aggregate token/cost stats across the whole trace |
| `EnvironmentContext` | Runtime info (framework, Python version, SDK version) |

### Domain Sections

Attach context relevant to the agent's domain via keyword arguments or `set_section()`:

| Section | Use Case |
|---------|----------|
| `Customer` | Customer name, email, company |
| `Sales` | Lead source, deal value, pipeline stage |
| `Support` | Ticket ID, issue type, resolution |
| `Research` | Topic, sources consulted, key findings |
| `Development` | Language, framework, test coverage |
| `Marketing` | Campaign, channel, target audience |
| `Content` | Content type, word count, SEO data |
| `Executive` | Meeting agenda, action items, decisions |
| `Project` | Deliverables, stakeholders, blockers |
| `General` | Tools used, tags, notes |

## Design Principle

The agent reports facts. Maestro evaluates. The SDK never asks the agent to score itself, say pass/fail, or rate its own confidence. It captures what happened -- timestamps, token counts, error types, what was produced -- and Maestro decides if the work was good.

## Documentation

For in-depth guides, API reference, and advanced usage, see the [full documentation](docs/).

## Development

```bash
git clone <repository-url>
cd infinium-python-sdk
pip install -e ".[dev]"
pytest
```

## Support

- Create an issue in the project repository
- Contact support at support@i42m.ai
