Metadata-Version: 2.4
Name: invokelens-sdk
Version: 0.2.3
Summary: SDK for InvokeLens — AI Agent Guardrails & Observability Platform
Author-email: InvokeLens <support@invokelens.com>
License: MIT
Project-URL: Homepage, https://invokelens.com
Project-URL: Repository, https://github.com/InvokeLens/invokelens-sdk
Project-URL: Changelog, https://github.com/InvokeLens/invokelens-sdk/blob/main/CHANGELOG.md
Project-URL: Documentation, https://docs.invokelens.com
Keywords: aws,bedrock,agents,observability,monitoring,guardrails,telemetry,llm
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Monitoring
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pydantic>=2.0
Requires-Dist: httpx>=0.24.0
Provides-Extra: eventbridge
Requires-Dist: boto3>=1.28.0; extra == "eventbridge"
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-asyncio; extra == "dev"
Requires-Dist: moto[dynamodb,events]>=4.0; extra == "dev"
Requires-Dist: mypy; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Dynamic: license-file

# InvokeLens SDK

**Observability & Guardrails for AWS Bedrock Agents**

InvokeLens captures telemetry from your AWS Bedrock agent invocations — cost, latency, token usage, tool calls, traces, and errors — and sends it to the InvokeLens platform for monitoring, alerting, and analysis.

## Installation

```bash
pip install invokelens-sdk
```

For EventBridge transport or Bedrock agent auto-resolution:

```bash
pip install invokelens-sdk[eventbridge]
```

## Quick Start

```python
from invokelens_sdk import InvokeLensClient

# Initialize the client
client = InvokeLensClient(api_key="il_live_abc123")

# Decorate your Bedrock agent function
@client.observe(
    agent_id="my-agent",
    agent_name="Customer Support Bot",
    bedrock_agent_id="ABCDEFGHIJ",  # auto-resolves model ID
)
def invoke_agent(prompt: str):
    import boto3
    bedrock = boto3.client("bedrock-agent-runtime")
    response = bedrock.invoke_agent(
        agentId="ABCDEFGHIJ",
        agentAliasId="TSTALIASID",
        sessionId="session-123",
        inputText=prompt,
    )
    return response

# Call your function as normal — telemetry is captured automatically
result = invoke_agent("What is the status of order #1234?")

# Flush remaining events on shutdown
client.shutdown()
```

## Client Configuration

| Parameter | Default | Description |
|-----------|---------|-------------|
| `api_key` | *(required)* | Your InvokeLens API key |
| `endpoint_url` | `https://api.invokelens.com` | InvokeLens ingest endpoint URL |
| `transport_mode` | `"http"` | Transport backend: `"http"` or `"eventbridge"` |
| `event_bus_name` | `None` | EventBridge bus name (required if transport_mode is `"eventbridge"`) |
| `batch_size` | `10` | Number of events per batch flush |
| `flush_interval` | `5.0` | Seconds between automatic flushes |
| `enable_kill_switch` | `True` | Enable pre-invocation kill switch and policy checks |
| `status_check_ttl` | `10.0` | TTL in seconds for cached agent status/policy checks |

## What Gets Captured

The `@client.observe()` decorator automatically captures:

- **Timing** — invocation start, end, and duration
- **Token usage** — input and output token counts (auto-detected from Bedrock response)
- **Model ID** — which Bedrock model was used (auto-detected)
- **Cost estimate** — computed from token usage and live model pricing
- **Status** — SUCCESS, FAILURE, or TIMEOUT
- **Errors** — exception type and message (truncated to 500 chars)
- **Tool calls** — names of tools invoked during execution
- **Traces** — nested span tree for the full invocation
- **Prompt fingerprint** — structural hash for drift detection
- **Prompt & response summaries** — truncated content for review

## `@client.observe()` Parameters

| Parameter | Default | Description |
|-----------|---------|-------------|
| `agent_id` | *(required)* | Unique identifier for the agent in InvokeLens |
| `agent_name` | `None` | Human-readable name for the agent |
| `model_id` | auto-detected | Bedrock model ID (e.g. `"anthropic.claude-3-sonnet"`) |
| `bedrock_agent_id` | `None` | Bedrock agent ID — enables auto model resolution via `GetAgent` |
| `bedrock_agent_alias_id` | `None` | Bedrock agent alias ID |
| `bedrock_region` | `AWS_DEFAULT_REGION` | AWS region for the `GetAgent` call |
| `boto3_session` | `None` | Optional `boto3.Session` for the `GetAgent` call (useful for local dev with named profiles) |
| `session_id` | `None` | Groups invocations into a logical conversation/workflow. Enables the Sessions tab in the dashboard |

## Model Detection

The SDK automatically resolves the model ID using this chain:

1. **Explicit override** — pass `model_id` directly to `@client.observe()`
2. **Response extraction** — reads `modelId` from the Bedrock response dict (works for `invoke_model`)
3. **Bedrock GetAgent API** — if you provide `bedrock_agent_id`, the SDK calls `GetAgent` once and caches the model (works for `invoke_agent`)
4. **Warning** — if none of the above produce a model ID, the SDK logs a warning and uses `"unknown"`

**Recommended for `invoke_agent` users** — pass your Bedrock agent ID:

```python
@client.observe(
    agent_id="my-agent",
    bedrock_agent_id="ABCDEFGHIJ",   # SDK auto-resolves model via GetAgent
)
def invoke_agent(prompt: str):
    ...
```

## Tracing

Every `@client.observe()` call creates a `TraceContext` with a root span. You can add nested spans for detailed invocation breakdowns.

### Automatic Trace Injection

If your decorated function accepts a `trace` keyword argument, the SDK injects the `TraceContext` automatically:

```python
from invokelens_sdk import TraceContext

@client.observe(agent_id="my-agent", model_id="anthropic.claude-3-sonnet")
def invoke_agent(prompt: str, trace: TraceContext = None):
    with trace.span("preprocess", span_type="custom") as s:
        processed = preprocess(prompt)
        s.output = processed

    with trace.span("call_llm", span_type="llm", model_id="anthropic.claude-3-sonnet") as s:
        result = call_bedrock(processed)
        s.output = result
        s.input_tokens = 500
        s.output_tokens = 200

    return result
```

### Tool Tracing

Use `@client.trace_tool()` to automatically create spans around tool function calls:

```python
@client.trace_tool(name="web_search")
def search(query: str, trace: TraceContext = None):
    return do_search(query)

@client.observe(agent_id="my-agent", model_id="anthropic.claude-3-sonnet")
def invoke_agent(prompt: str, trace: TraceContext = None):
    results = search(query=prompt, trace=trace)
    return results
```

### Span Types

| Type | Description |
|------|-------------|
| `llm` | LLM/model invocation |
| `tool` | Tool or action group call |
| `chain` | Multi-step chain (root span default) |
| `retrieval` | RAG/knowledge base retrieval |
| `guardrail` | Guardrail evaluation |
| `custom` | Any custom step |

## Prompt Fingerprinting

The SDK computes a structural fingerprint of each prompt for drift detection:

```python
from invokelens_sdk import compute_fingerprint

fp = compute_fingerprint("Summarize the following: {document}")
# {
#     "prompt_hash": "sha256...",       # Hash of full normalized text
#     "structure_hash": "sha256...",    # Hash with template vars replaced
#     "char_count": 38,
#     "word_count": 5,
#     "line_count": 1,
#     "template_vars": ["document"],
# }
```

Fingerprints are automatically included in telemetry events. The InvokeLens dashboard uses them to detect prompt drift between deployments.

## Guardrails & Kill Switch

The SDK enforces guardrail policies before each invocation. Policies are configured in the InvokeLens dashboard and cached locally with a configurable TTL (default: 10s).

### Kill Switch

If an agent is marked as **BLOCKED** in the dashboard, the SDK raises `AgentBlockedError` *before* making any Bedrock call (zero cost incurred):

```python
from invokelens_sdk import AgentBlockedError

try:
    result = invoke_agent("Hello")
except AgentBlockedError as e:
    print(f"Agent blocked: {e}")
```

To disable the kill switch:

```python
client = InvokeLensClient(api_key="...", enable_kill_switch=False)
```

### Policy Types

| Policy | Description |
|--------|-------------|
| `COST_CAP` | Blocks if estimated invocation cost exceeds a USD threshold |
| `TOKEN_LIMIT` | Blocks if estimated input tokens exceed a limit |
| `RATE_LIMIT` | Blocks if invocation count exceeds N per sliding time window |
| `TIME_RESTRICTION` | Blocks invocations outside allowed UTC hours |

Policies with `enforcement: "BLOCK"` raise `PolicyViolationError`. Policies with `enforcement: "LOG"` are recorded but do not block.

```python
from invokelens_sdk import PolicyViolationError

try:
    result = invoke_agent("Hello")
except PolicyViolationError as e:
    print(f"Policy violated: {e.policy_type} — {e.violation_message}")
```

## Cost Estimation

The SDK estimates invocation costs using live pricing from the AWS Pricing API (fetched once, cached for process lifetime), with hardcoded fallbacks when the API is unavailable.

```python
from invokelens_sdk import estimate_cost

cost = estimate_cost(
    model_id="anthropic.claude-3-sonnet",
    input_tokens=1000,
    output_tokens=500,
)
print(f"Estimated cost: ${cost:.4f}")
```

### Custom Pricing

Override pricing for models not yet in the built-in list:

```python
from invokelens_sdk import set_custom_pricing

set_custom_pricing("my-custom-model", input_per_1k=0.001, output_per_1k=0.002)
```

### Supported Models

Nova (Micro, Lite, Pro, 2.0), Claude (3 Haiku/Sonnet/Opus, 3.5 Haiku/Sonnet, 3.7 Sonnet, 4 Sonnet/Opus), Titan Text (Lite, Express, Premier), Llama 3/3.1, Mistral (Small, Large), Cohere Command R/R+.

## Transport Modes

### HTTP (Default)

Sends batched events to the InvokeLens API via HTTPS. Includes automatic retry with exponential backoff (up to 3 retries). Handles quota exceeded (HTTP 429) gracefully — stops queueing events and logs an actionable upgrade message.

### EventBridge

Publishes events to an Amazon EventBridge bus. Useful for AWS-native architectures where you want to process events through EventBridge rules.

```python
client = InvokeLensClient(
    api_key="il_live_abc123",
    transport_mode="eventbridge",
    event_bus_name="invokelens-events",
)
```

Requires `boto3`: `pip install invokelens-sdk[eventbridge]`

## Requirements

- Python 3.11+
- `pydantic` >= 2.0
- `httpx` >= 0.24.0
- `boto3` >= 1.28.0 *(optional — needed for EventBridge transport, `bedrock_agent_id` auto-resolution, and live pricing)*

## License

MIT License. See [LICENSE](LICENSE) for details.
