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.2.tar.gz (23.6 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.2-py3-none-any.whl (26.4 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: zentaxa-1.1.2.tar.gz
  • Upload date:
  • Size: 23.6 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.2.tar.gz
Algorithm Hash digest
SHA256 f6847ddcf4aaa70a5a02be363ab61d26908d5d7443418067df82d522cf8a4820
MD5 cfab3de753fa5349b8bbb2772d0c9da0
BLAKE2b-256 acd066cd212fce0dcdaa4eadd7305771508768fd9e0f75babe06f9a5f3ded277

See more details on using hashes here.

File details

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

File metadata

  • Download URL: zentaxa-1.1.2-py3-none-any.whl
  • Upload date:
  • Size: 26.4 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.2-py3-none-any.whl
Algorithm Hash digest
SHA256 1ef278305e458dcc44b2d84554354b118e79b736798a0269a66b04f83606fe76
MD5 19f595ff7aa2b1c0a032cc2567075324
BLAKE2b-256 e78fb59696edada0ff8fdb198a3c50e4f1deb5cf6da3e16354f2c05bf6bedf32

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