Python SDK

pip install tenzro-cloud

from tenzro_cloud import Tenzro
import os

client = Tenzro(
    api_key=os.environ["TENZRO_API_KEY"],
    project_id=os.environ.get("TENZRO_PROJECT_ID"),  # optional default
)

# Simple chat - just provide a prompt!
response = client.ai.chat("What is machine learning?")
print(response)

from tenzro_cloud import AsyncTenzro
import asyncio

async def main():
    client = AsyncTenzro(api_key=os.environ["TENZRO_API_KEY"])

    response = await client.ai.chat("Hello!")
    print(response)

    await client.close()

asyncio.run(main())

# Simple chat
response = client.ai.chat("Explain quantum computing")

# With options
response = client.ai.chat(
    "Write a haiku",
    provider="anthropic",
    model="claude-sonnet-4",
    temperature=0.9,
)

# Full control with infer()
result = client.ai.infer(
    provider="google",
    model="gemini-2.5-pro",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello!"},
    ],
    temperature=0.7,
    max_tokens=1024,
)
print(result.content)
print(f"Tokens used: {result.usage.total_tokens}")

# Get database by name
vectors = client.vec.db("embeddings")

# Insert documents - embeddings are auto-generated!
vectors.upsert(
    documents=[
        {"id": "doc-1", "text": "Machine learning transforms data"},
        {"id": "doc-2", "text": "Neural networks are powerful", "metadata": {"category": "ai"}},
    ]
)

# Query with natural language
results = vectors.query(
    text="How does AI learn?",
    top_k=5,
)

for match in results:
    print(f"{match.metadata['_text']} - score: {match.score}")

# Create a new database
db = client.vec.create_database(
    db_name="my-embeddings",
    dimension=768,
    metric_type="COSINE",
)

# Get store by name
store = client.kev.store("sessions")

# Set value with TTL
store.set("user:123", {"name": "John", "role": "admin"}, ttl=3600)

# Get value
user = store.get("user:123")

# Hash operations
store.hset("user:123", "preferences", {"theme": "dark"})
prefs = store.hget("user:123", "preferences")

# List operations
store.lpush("queue", "task-1", "task-2")
task = store.rpop("queue")

from tenzro_cloud import tool

# Define custom tools
search_tool = tool(
    name="search_products",
    description="Search the product catalog",
    properties={
        "query": {"type": "string", "description": "Search query"},
        "category": {"type": "string", "description": "Product category"},
    },
    required=["query"],
)

# Create an agent
agent = client.agents.create(
    agent_name="Shopping Assistant",
    endpoint_id="your-endpoint-id",
    system_prompt="You help customers find products.",
    orchestration_pattern="SINGLE",
    tools=[search_tool],
)

# Activate the agent
client.agents.activate(agent.agent_id)

# Chat with the agent
response = client.agents.chat(
    agent_id=agent.agent_id,
    message="Find me wireless headphones",
)
print(response.content)

# Stream responses
for chunk in client.ai.chat_stream("Write a story about AI"):
    print(chunk, end="", flush=True)

# Or with full control
stream = client.ai.infer_stream(
    provider="google",
    model="gemini-2.5-flash",
    prompt="Write a poem about clouds",
)

for chunk in stream:
    print(chunk.content or "", end="", flush=True)

# Automatically close client
with Tenzro(api_key=api_key) as client:
    response = client.ai.chat("Hello!")

# Async context manager
async with AsyncTenzro(api_key=api_key) as client:
    response = await client.ai.chat("Hello!")

from tenzro_cloud import TenzroError, AuthenticationError, RateLimitError

try:
    client.ai.chat("Hello")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited, retry after: {e.retry_after}")
except TenzroError as e:
    print(f"API error {e.status_code}: {e.message}")