yantrix 1.0.0

Python SDK for Yantrix — agent infrastructure with governance built in

yantrix

Python SDK for Yantrix — agent infrastructure with governance built in.

Install

pip install yantrix

Quickstart

import yantrix

# Configure once
yantrix.configure(
    api_key="ytx_your_key_here",
    agent_id="my_agent"
)

# Run anything — policy check + memory + execution + audit, automatically
result = yantrix.run(
    intent="research openai pricing changes",
    task={
        "steps": [
            {
                "name": "fetch",
                "type": "http",
                "url": "https://gateway.yantrix.ai/proxy/compwatch/competitor/scan",
                "method": "POST",
                "body": {"company": "openai"},
                "output_key": "intel"
            }
        ]
    }
)

print(result["trace_id"])    # ytx_run_abc123
print(result["output"])      # {"intel": {...}}
print(result["timeline"])    # [{layer: "policy", ...}, {layer: "memory", ...}, ...]

What happens on every .run() call

1. Policy Engine evaluates the action  → is this allowed?
2. Memory searched for context         → what does agent already know?
3. Execution routed automatically      → direct | sequential | graph
4. Result stored in semantic memory    → agent learns from this
5. Full trace returned                 → audit trail included

Client API

from yantrix import Yantrix

ytx = Yantrix(api_key="ytx_...")

# Runtime
result = ytx.run(intent="...", task={...})
trace  = ytx.get_trace(result["trace_id"])

# Policy Engine
ytx.set_budget(daily_limit_usd=10.0, per_action_limit_usd=1.0)
spend  = ytx.get_spend()
policy = ytx.evaluate_policy("data_call", "web_crawler.fetch", estimated_cost=0.005)
logs   = ytx.get_audit_logs(limit=20)

# Memory
ytx.remember(key="user_pref", value="dark mode", memory_type="semantic")
results = ytx.recall(query="what does the user prefer?", top_k=5)
ytx.forget(key="user_pref")

# Flows
flow   = ytx.create_flow(name="my_flow", steps=[...])
output = ytx.run_flow(flow["flow_id"], input_data={})

# Graphs
graph  = ytx.create_graph(name="my_graph", nodes=[...], edges=[...])
output = ytx.run_graph(graph["graph_id"], input_data={})

Async support

import asyncio
from yantrix import Yantrix

ytx = Yantrix(api_key="ytx_...")

async def main():
    result = await ytx.arun(intent="...", task={...})
    print(result)

asyncio.run(main())

Environment variables

export YANTRIX_API_KEY=ytx_your_key_here
export YANTRIX_AGENT_ID=my_agent

Then use without configure():

import yantrix
result = yantrix.run(intent="...")

Error handling

from yantrix import Yantrix
from yantrix.client import PolicyDeniedError, YantrixError

ytx = Yantrix(api_key="ytx_...")

try:
    result = ytx.run(intent="expensive operation", task={...})
except PolicyDeniedError as e:
    print(f"Blocked by policy: {e.reason}")
    print(f"Audit ID: {e.audit_id}")
except YantrixError as e:
    print(f"Error {e.status_code}: {e}")

Links

• Yantrix Platform
• Documentation
• API Reference
• GitHub