Skip to main content

Framework-agnostic observability SDK for multi-agent workflows

Project description

ZENTAXA Python SDK

Framework-agnostic observability SDK for multi-agent workflows.

🚀 Live Platform: https://zentaxaapp.azurewebsites.net

Features

  • 🔍 Universal Telemetry: Unified API for all major frameworks
  • 🤖 5 Framework Integrations: LangChain, LangGraph, CrewAI, AutoGen, LlamaIndex
  • 📊 Real-time Tracking: Agent runs, steps, LLM calls, tools
  • 🔄 Multi-Step Visualization: Track and visualize complex agent pipelines
  • 🚀 Easy Integration: Add 2 lines of code to your existing agents
  • Async Support: Non-blocking telemetry capture
  • 🛡️ Error Handling: Automatic retry logic with exponential backoff

Installation

pip install zentaxa

Framework-specific installation

# LangChain / LangGraph
pip install zentaxa[langchain]

# CrewAI
pip install zentaxa[crewai]

# AutoGen
pip install zentaxa[autogen]

# LlamaIndex
pip install zentaxa[llamaindex]

# All frameworks
pip install zentaxa[all]

Quick Start

Basic Usage

from zentaxa import ZentaxaClient

# Default: connects to https://zentaxaapp.azurewebsites.net
client = ZentaxaClient()

# Start an agent run
response = client.trace(
    run_id=None,
    agent_id="my-agent",
    framework="custom",
    event_type="agent_start",
    metadata={"input": "Your query here"}
)
run_id = response["run_id"]

# Log steps
client.log(run_id, "info", "Processing step 1")

# End the run with results
client.trace(
    run_id=run_id,
    agent_id="my-agent",
    framework="custom",
    event_type="agent_end",
    metadata={
        "input": "Your query",
        "output": "Agent response",
        "steps": [
            {"step_number": 1, "step_name": "Analyze", "input": "...", "output": "...", "latency_ms": 1500, "status": "completed"}
        ],
        "latency_ms": 1500
    }
)

LangChain

from zentaxa import ZentaxaClient
from zentaxa.integrations.langchain import ZentaxaCallbackHandler
from langchain_openai import ChatOpenAI

# Initialize client (default connects to live platform)
client = ZentaxaClient()

# Create callback handler
handler = ZentaxaCallbackHandler(client=client, agent_id="my-agent")

# Use with LangChain
llm = ChatOpenAI(callbacks=[handler])
result = llm.invoke("What is quantum computing?")

LangGraph

from zentaxa.integrations.langgraph import LangGraphObserver
from langgraph.graph import StateGraph

observer = LangGraphObserver(client=client, agent_id="my-graph")

graph = StateGraph(state_schema)
graph.add_node("research", research_node, callbacks=[observer])
graph.add_node("analyze", analyze_node, callbacks=[observer])
compiled = graph.compile()

result = compiled.invoke({"query": "quantum computing"})

CrewAI

from zentaxa.integrations.crewai import CrewAIObserver
from crewai import Crew

observer = CrewAIObserver(client=client, agent_id="research-crew")

# Context manager automatically tracks execution
with observer:
    result = crew.kickoff()

AutoGen

from zentaxa.integrations.autogen import AutoGenTracer
from autogen import UserProxyAgent, AssistantAgent

tracer = AutoGenTracer(client=client, agent_id="autogen-conversation")

with tracer:
    user_proxy.initiate_chat(assistant, message="Research quantum computing")

LlamaIndex

from zentaxa.integrations.llamaindex import LlamaIndexObserver
from llama_index.core.agent import ReActAgent
from llama_index.core.callbacks import CallbackManager

observer = LlamaIndexObserver(client=client, agent_id="llama-agent")
callback_manager = CallbackManager([observer])

agent = ReActAgent.from_tools(
    tools,
    llm=llm,
    callback_manager=callback_manager
)

response = agent.chat("Research quantum computing")

Configuration

Client Options

client = ZentaxaClient(
    base_url="https://zentaxaapp.azurewebsites.net",  # Default (live platform)
    timeout=10.0,                                      # Request timeout (seconds)
    max_retries=3,                                     # Retry attempts
    async_mode=False                                   # Enable async operations
)

Async Mode

import asyncio

async def main():
    async with ZentaxaClient(async_mode=True) as client:
        await client.trace_async(
            run_id=None,
            agent_id="async-agent",
            framework="langchain",
            event_type="agent_start"
        )

asyncio.run(main())

Manual Telemetry

Trace Events

# Start agent run
response = client.trace(
    run_id=None,
    agent_id="research-agent",
    framework="langchain",
    event_type="agent_start",
    metadata={"goal": "Research quantum computing"}
)
run_id = response["run_id"]

# End agent run
client.trace(
    run_id=run_id,
    agent_id="research-agent",
    framework="langchain",
    event_type="agent_end",
    metadata={"output": "Research complete"}
)

Log Events

client.log(
    run_id=run_id,
    message="Executing search tool",
    level="info",
    context={"tool_name": "TavilySearch", "input": "quantum computing"}
)

Metric Events

client.metric(
    run_id=run_id,
    metric_name="llm_latency_ms",
    value=2340.5,
    tags={"provider": "openai", "model": "gpt-4"}
)

Agent Events

client.agent_event(
    run_id=run_id,
    event_type="step_complete",
    step_number=1,
    step_name="Plan Research",
    action_type="plan",
    input_data="Research quantum computing",
    output_data="1. Search papers\n2. Analyze trends",
    latency_ms=150,
    cost_usd=0.0001
)

API Reference

ZentaxaClient

  • trace(run_id, agent_id, framework, event_type, metadata) - Send trace event
  • log(run_id, message, level, context) - Send log event
  • metric(run_id, metric_name, value, tags) - Send metric event
  • agent_event(run_id, event_type, ...) - Send agent event
  • bulk(events) - Send multiple events in batch

Callback Handlers

  • ZentaxaCallbackHandler - LangChain/LangGraph callback
  • LangGraphObserver - LangGraph-specific observer
  • CrewAIObserver - CrewAI wrapper
  • AutoGenTracer - AutoGen conversation tracker
  • LlamaIndexObserver - LlamaIndex event handler

Development

# Clone repository
git clone https://github.com/zentaxa/zentaxa-sdk

# Install dependencies
cd zentaxa-sdk/sdk/python
pip install -e ".[dev]"

# Run tests
pytest

# Format code
black zentaxa/

License

MIT License - see LICENSE file for details.

Support

Version

1.0.1

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

zentaxa-1.1.1.tar.gz (23.3 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

zentaxa-1.1.1-py3-none-any.whl (26.0 kB view details)

Uploaded Python 3

File details

Details for the file zentaxa-1.1.1.tar.gz.

File metadata

  • Download URL: zentaxa-1.1.1.tar.gz
  • Upload date:
  • Size: 23.3 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.10.12

File hashes

Hashes for zentaxa-1.1.1.tar.gz
Algorithm Hash digest
SHA256 432007b13403b0ea336c76d52f4e52208829fc615227cfd0cd3034ba9c412ee9
MD5 bbd9d8f428b2ea1c1dea27d44008ef83
BLAKE2b-256 7a0b94b88240bbd23446c73bd7f473a943c1fb21810fa9b5330887d3a9be7fbb

See more details on using hashes here.

File details

Details for the file zentaxa-1.1.1-py3-none-any.whl.

File metadata

  • Download URL: zentaxa-1.1.1-py3-none-any.whl
  • Upload date:
  • Size: 26.0 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.10.12

File hashes

Hashes for zentaxa-1.1.1-py3-none-any.whl
Algorithm Hash digest
SHA256 1ceb9e58ac4fe0149031d2f209765b2ed4df2f52044b5549bfcdc9896477602f
MD5 d3fc5e934445305436a9cd9e32187ce1
BLAKE2b-256 3f170a4f1ec0a30357f20bba7810640723431072339faea7dabc10b25b2ae55e

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page