Metadata-Version: 2.4
Name: goantiai-sdk
Version: 1.0.0
Summary: Official Python SDK for Oculus, Identity & Access management Platform for AI Agents by Anti AI
Author: Anti AI
License-Expression: MIT
Keywords: ai,agent,iam,oauth,sdk,langchain,openai,crewai
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
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 :: Security
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.24.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: PyJWT>=2.8.0
Requires-Dist: cryptography>=41.0.0
Provides-Extra: langchain
Requires-Dist: langchain>=0.1.0; extra == "langchain"
Requires-Dist: langchain-core>=0.1.0; extra == "langchain"
Provides-Extra: crewai
Requires-Dist: crewai>=0.1.0; extra == "crewai"
Provides-Extra: openai
Requires-Dist: openai>=1.0.0; extra == "openai"
Provides-Extra: yaml
Requires-Dist: PyYAML>=6.0; extra == "yaml"
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21; extra == "dev"
Requires-Dist: pytest-cov>=4.0; extra == "dev"

# goantiai-sdk

Official Python SDK for Oculus, Identity & Access management Platform for AI Agents by [Anti AI](https://goantiai.com).

Enable your AI agents to request scoped credentials, enforce capability policies at runtime, delegate to child agents, and integrate with LangChain, CrewAI, or OpenAI — with built-in audit logging and security enforcement.

## Installation

```bash
pip install goantiai-sdk
```

With framework integrations:

```bash
pip install goantiai-sdk[langchain]
pip install goantiai-sdk[crewai]
pip install goantiai-sdk[openai]
pip install goantiai-sdk[yaml]  # for oculus.yaml custom capabilities
```

## Quick Start

```python
from oculus_sdk import OculusClient

client = OculusClient(
    client_id="agt_abc123",
    client_secret="cs_live_xxx",
    base_url="https://api.goantiai.com",
    task_id="my-task-001",
)
client.startup()

@client.tool(capability="stripe.customers.read")
def get_customer(customer_id: str) -> dict:
    import stripe
    stripe.api_key = client.credentials["stripe"]["stripe_secret_key"]
    return stripe.Customer.retrieve(customer_id)

# Wrap tools with enforcement — now every call is checked against the agent's policy
tools = client.wrap_tools([get_customer])
```

The `startup()` call handles:
- Token acquisition (client credentials flow)
- JWKS key fetching for local token validation
- Capability map sync
- Token revocation list sync
- Credential fetching for configured integrations
- Loading custom capabilities from `oculus.yaml`
- Starting background refresh threads

## Configuration

```python
client = OculusClient(
    client_id="agt_abc123",        # or set OCULUS_CLIENT_ID env var
    client_secret="cs_live_xxx",   # or set OCULUS_CLIENT_SECRET env var
    base_url="https://api.goantiai.com",  # or set OCULUS_BASE_URL env var
    task_id="task-123",            # optional, or set OCULUS_TASK_ID env var
    scopes=["stripe.customers.read", "email:send"],  # requested scopes
    fail_closed=True,              # default: True — raises on startup failures
)
```

## Tool Wrapping with Capability Enforcement

The core feature: decorate functions with their required capability, then wrap them. Every call is checked against the agent's granted scopes and policy constraints before execution.

```python
@client.tool(capability="stripe.customers.read")
def get_customer(customer_id: str) -> dict:
    stripe.api_key = client.credentials["stripe"]["stripe_secret_key"]
    return stripe.Customer.retrieve(customer_id)

@client.tool(capability="stripe.charges.create")
def create_charge(amount: int, currency: str, customer_id: str) -> dict:
    return stripe.Charge.create(amount=amount, currency=currency, customer=customer_id)

# Wrap all tools at once
tools = client.wrap_tools([get_customer, create_charge])

# Now calling tools[0]("cus_123") will:
# 1. Check token validity (expiry + revocation)
# 2. Check capability against policy (scope + constraints)
# 3. Execute the function
# 4. Emit an audit event (non-blocking)
# Raises PolicyViolationError if denied.
```

## Task Context (Framework-Agnostic)

For code that doesn't use LangChain/CrewAI/OpenAI, use the context manager pattern:

```python
with client.task("refund-batch-001") as ctx:
    # Non-raising check
    if ctx.can("stripe.charges.create"):
        creds = ctx.credentials("stripe")
        stripe.api_key = creds["stripe_secret_key"]
        # do the thing...

    # Raising check with automatic credential injection
    with ctx.capability("stripe.customers.read") as creds:
        stripe.api_key = creds["stripe_secret_key"]
        customer = stripe.Customer.retrieve("cus_123")
```

- `ctx.can(capability)` — returns `True`/`False`, never raises
- `ctx.capability(capability)` — context manager that raises `PolicyViolationError` if denied, yields credentials
- `ctx.credentials(integration)` — returns credential dict, raises `CredentialError` if not found

## Delegation (Agent-to-Agent)

A parent agent can delegate a subset of its capabilities to a child agent:

```python
child_token = client.delegate(
    child_agent_id="agent-worker-001",
    capabilities=["stripe.customers.read"],  # must be subset of parent's scopes
    task_id="subtask-456",
    ttl_seconds=300,
    constraints={"max_amount_cents": 5000},  # can only tighten, not loosen
)
```

### Child Agent Using a Delegation Token

```python
child_client = OculusClient.from_delegation_token(
    token=child_token,
    base_url="https://api.goantiai.com",
)

# Child can only use the delegated capabilities
tools = child_client.wrap_tools([get_customer])
```

## Accessing Credentials

After `startup()`, credentials for configured integrations are available:

```python
stripe_key = client.credentials["stripe"]["stripe_secret_key"]
github_token = client.credentials["github"]["access_token"]
```

## Token Management (v1 API)

For simpler use cases — request a scoped token directly:

```python
# Sync
token = client.get_token("stripe.customers.read email:send")

# Async
token = await client.aget_token("stripe.customers.read", audience="https://api.stripe.com")
```

Tokens are automatically cached, refreshed, and checked against the revocation list.

## LangChain Integration

```python
from oculus_sdk.langchain import OculusLangChain

client = OculusLangChain(
    client_id="agt_abc123",
    client_secret="cs_live_xxx",
    base_url="https://api.goantiai.com",
    task_id="my-task",
)
client.startup()

@client.tool(capability="stripe.customers.read")
def get_customer(customer_id: str) -> str:
    """Retrieve a Stripe customer by ID."""
    stripe.api_key = client.credentials["stripe"]["stripe_secret_key"]
    return str(stripe.Customer.retrieve(customer_id))

# Returns LangChain StructuredTool objects
tools = client.wrap_tools([get_customer])

# Use the callback handler to capture LLM reasoning for audit
from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    callbacks=[client.callback_handler],  # captures reasoning for audit
)
agent.run("Look up customer cus_123")
```

If a capability is denied, the tool returns an error to the LLM so the agent can handle it gracefully.

## CrewAI Integration

```python
from oculus_sdk.crewai import OculusCrew

client = OculusCrew(
    client_id="agt_abc123",
    client_secret="cs_live_xxx",
    base_url="https://api.goantiai.com",
    task_id="my-task",
)
client.startup()

@client.tool(capability="stripe.customers.read")
def get_customer(customer_id: str) -> str:
    """Retrieve a Stripe customer."""
    stripe.api_key = client.credentials["stripe"]["stripe_secret_key"]
    return str(stripe.Customer.retrieve(customer_id))

# Option 1: Wrap individual tools
wrapped = client.wrap_tool(get_customer, capability="stripe.customers.read")

# Option 2: Wrap an entire crew at once
from crewai import Agent, Crew, Task

researcher = Agent(role="Researcher", tools=[get_customer], ...)
crew = Crew(agents=[researcher], tasks=[...])
secured_crew = client.wrap_crew(crew)  # wraps all tools across all agents
secured_crew.kickoff()
```

## OpenAI Integration

```python
from oculus_sdk.openai import OculusOpenAI

client = OculusOpenAI(
    client_id="agt_abc123",
    client_secret="cs_live_xxx",
    base_url="https://api.goantiai.com",
    task_id="my-task",
)
client.startup()

# Register function → capability mappings
client.register_function("get_customer", capability="stripe.customers.read")
client.register_function("create_charge", capability="stripe.charges.create")

# Your function implementations
registry = {
    "get_customer": lambda customer_id: stripe.Customer.retrieve(customer_id),
    "create_charge": lambda amount, currency, customer: stripe.Charge.create(...),
}

# Use OpenAI normally
response = client.openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Look up customer cus_123"}],
    tools=[...],
)

# Execute tool calls with enforcement
for tool_call in response.choices[0].message.tool_calls or []:
    result = client.execute_tool_call(tool_call, registry)

# Or process all at once (returns tool message dicts for the conversation)
results = client.execute_tool_calls(
    response.choices[0].message.tool_calls or [], registry
)
```

## Capability Evaluator

The SDK includes an offline evaluator that checks capabilities against constraints without network calls:

```python
from oculus_sdk import CapabilityEvaluator

evaluator = CapabilityEvaluator()

result = evaluator.evaluate(
    capability="stripe.charges.create",
    token_scopes={"stripe.charges.create", "stripe.customers.read"},
    constraints={
        "max_amount_cents": 10000,
        "time_bounds": {
            "days": ["mon", "tue", "wed", "thu", "fri"],
            "hours": "09:00-17:00",
            "timezone": "America/New_York",
        },
        "task_context_required": True,
    },
    context={"task_id": "task-123", "amount_cents": 5000},
)

print(result.allowed)  # True or False
print(result.reason)   # "allowed", "capability_not_in_scope", "time_window_violation", etc.
```

### Constraint Types

| Constraint | Description |
|-----------|-------------|
| `time_bounds` | Restrict to specific days and hours (with timezone) |
| `max_amount_cents` | Cap transaction amounts |
| `task_context_required` | Require a task_id to be present |
| `max_results` | Limit number of results returned |

## Custom Capabilities (oculus.yaml)

Define custom capabilities in an `oculus.yaml` file in your project root:

```yaml
custom_capabilities:
  internal.reports.generate:
    description: "Generate internal reports"
    risk_level: "medium"
    tools: ["generate_report"]
  internal.data.export:
    description: "Export data to CSV"
    risk_level: "high"
    tools: ["export_csv"]
```

These are loaded automatically during `startup()` if `PyYAML` is installed (`pip install goantiai-sdk[yaml]`).

## Token Verification (For Downstream Services)

If you're building a service that receives tokens from agents and needs to verify them:

```python
from oculus_sdk import JwksVerifier

verifier = JwksVerifier(base_url="https://api.goantiai.com")

payload = await verifier.verify(token_from_agent)
print(payload["sub"])        # agent ID
print(payload["scope"])      # granted scopes
print(payload["tenant_id"])  # tenant
```

The verifier fetches and caches JWKS public keys, validates RS256 signatures, and checks the token revocation list.

## Agent Enrollment (v1 API)

For first-time agent registration using a one-time enroll token:

```python
client = OculusClient.from_enroll_token(
    enroll_token="enroll_xxx",
    base_url="https://api.goantiai.com",
)
# client now has client_id and client_secret populated
```

## Lambda Cold-Start Optimization

For AWS Lambda deployments, enable caching to avoid re-fetching on warm starts:

```bash
export OCULUS_LAMBDA_CACHE=true
```

The SDK caches token, JWKS, and capability maps to `/tmp/oculus_cache.json` and restores on subsequent invocations if the token is still valid.

## Error Handling

```python
from oculus_sdk import (
    PolicyViolationError,
    TokenExpiredError,
    TokenRevokedException,
    OculusAuthError,
    OculusAPIError,
    OculusStartupError,
    DelegationError,
    CredentialError,
)

try:
    result = get_customer("cus_123")
except PolicyViolationError as e:
    print(e.capability)           # which capability was denied
    print(e.reason)               # why (e.g. "capability_not_in_scope")
    print(e.granted_capabilities) # what the token actually has
    print(e.policy_id)            # policy that made the decision
except TokenExpiredError:
    client.startup()              # re-initialize
except TokenRevokedException as e:
    print(e.jti)                  # revoked token ID
except DelegationError as e:
    print(e.reason)               # why delegation failed
except CredentialError as e:
    print(e.integration)          # which integration is missing credentials
```

## Audit

Audit events are emitted automatically when using `wrap_tools()`, `TaskContext`, or framework integrations. Events include:
- Capability checked
- Decision (allow/deny) and reason
- Tool name and arguments (credentials auto-redacted)
- Agent reasoning (when using LangChain callback handler)
- Task ID, delegation depth, framework

Events are batched and sent to the control plane every 5 seconds. Audit is best-effort — it never blocks or delays tool execution.

## Cleanup

Stop background threads when your agent is done:

```python
client.stop()
```

## Environment Variables

| Variable | Description |
|----------|-------------|
| `OCULUS_CLIENT_ID` | Agent client ID |
| `OCULUS_CLIENT_SECRET` | Agent client secret |
| `OCULUS_BASE_URL` | API base URL |
| `OCULUS_TASK_ID` | Current task identifier |
| `OCULUS_LAMBDA_CACHE` | Set to `true` for Lambda cold-start optimization |

## Requirements

- Python 3.9+

## License

MIT
