goantiai-sdk 1.0.0

Official Python SDK for Oculus, Identity & Access management Platform for AI Agents by Anti AI

goantiai-sdk

Official Python SDK for Oculus, Identity & Access management Platform for AI Agents by Anti AI.

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

pip install goantiai-sdk

With framework integrations:

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

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

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.

@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:

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:

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

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:

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:

# 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

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

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

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:

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:

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:

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:

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:

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

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:

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