vyrt 0.1.0

Python SDK for the VYRT observability platform

vyrt-python

Python SDK for the VYRT observability platform.

Installation

pip install vyrt

Core client

from vyrt import VyrtClient

client = VyrtClient(
    api_key="vk_live_...",
    base_url="https://your-vyrt-instance.com",  # optional, defaults to http://localhost:3000
)

# Ingest a single run
client.ingest({"model": "gpt-4o", "tokens": 512, "latency_ms": 320})

# Ingest multiple runs in one request
client.batch([
    {"model": "gpt-4o", "tokens": 200},
    {"model": "claude-3-5-sonnet", "tokens": 150},
])

# Send an agent heartbeat
client.heartbeat({"agentId": "worker-1", "status": "running", "queue_depth": 4})

All methods retry up to 3 times with exponential backoff on network errors or 5xx responses via urllib3.Retry.

---

Integrations

LangChain — VyrtCallbackHandler

Requires: pip install langchain-core

Attach to any chain or agent via the callbacks config key. Captures on_chain_start, on_chain_end, on_chain_error, on_tool_start, and on_tool_end, then calls client.ingest when the chain completes.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from vyrt import VyrtClient, VyrtCallbackHandler

client = VyrtClient(api_key="vk_live_...")
handler = VyrtCallbackHandler(client, metadata={"environment": "production", "user_id": "u_123"})

chain = (
    ChatPromptTemplate.from_template("Summarise: {text}")
    | ChatOpenAI(model="gpt-4o")
    | StrOutputParser()
)

result = chain.invoke(
    {"text": "LangChain is a framework for building LLM apps."},
    config={"callbacks": [handler]},
)

Captured events

LangChain callback	Vyrt action
on_chain_start	Opens a run
on_tool_start	Appends a tool step
on_tool_end	Closes the tool step
on_chain_end	Closes run → client.ingest with status: "success"
on_chain_error	Closes run → client.ingest with status: "error"

---

CrewAI — VyrtCrewObserver

Requires: pip install crewai

observer.wrap(crew) attaches listeners to the CrewAI event bus for TaskStartedEvent, TaskCompletedEvent, and TaskFailedEvent. Returns the same crew instance for chaining.

from crewai import Agent, Crew, Task
from vyrt import VyrtClient, VyrtCrewObserver

client = VyrtClient(api_key="vk_live_...")
observer = VyrtCrewObserver(client, metadata={"project": "research-bot"})

researcher = Agent(
    role="Researcher",
    goal="Find accurate information",
    backstory="Expert at internet research",
    llm="gpt-4o",
)

task = Task(
    description="Research the history of the Eiffel Tower",
    expected_output="A concise summary with key dates",
    agent=researcher,
)

crew = observer.wrap(Crew(agents=[researcher], tasks=[task]))
result = crew.kickoff()

Captured fields per ingest

• task_id, description, expected_output
• agent.role, agent.goal, agent.backstory
• status ("success" | "error"), started_at, ended_at
• output, error

---

LlamaIndex — VyrtLlamaObserver

Requires: pip install llama-index-core

Register with LlamaIndex's CallbackManager. Captures query, llm, and tool event pairs and sends the completed run on query end.

from llama_index.core import Settings, VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.callbacks import CallbackManager
from vyrt import VyrtClient, VyrtLlamaObserver

client = VyrtClient(api_key="vk_live_...")
observer = VyrtLlamaObserver(client, metadata={"index_name": "docs"})

Settings.callback_manager = CallbackManager([observer])

documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

response = query_engine.query("What is LlamaIndex?")

Captured events

LlamaIndex event	Vyrt action
query start	Opens a run
llm start	Appends an llm step
llm end	Closes the llm step
tool start	Appends a tool step
tool end	Closes the tool step
query end	Closes run → client.ingest

---

Custom Agent — VyrtTracer

For teams not using a framework. Full manual control over run and step lifecycle.

from vyrt import VyrtClient, VyrtTracer

client = VyrtClient(api_key="vk_live_...")
tracer = VyrtTracer(client, metadata={"service": "my-agent"})


def run_pipeline(query: str) -> str | None:
    run_id = tracer.start_run("pipeline", metadata={"query": query})

    # Step 1 — retrieval
    step_id = tracer.add_step(run_id, "retrieve_context", step_type="retrieval", input=query)
    try:
        docs = fetch_relevant_docs(query)
        tracer.end_step(run_id, step_id, output=docs)
    except Exception as exc:
        tracer.fail_step(run_id, step_id, exc)
        tracer.fail_run(run_id, exc)
        return None

    # Step 2 — LLM call
    step_id = tracer.add_step(run_id, "generate_answer", step_type="llm", input=query)
    try:
        answer = call_llm(query, docs)
        tracer.end_step(run_id, step_id, output=answer)
        tracer.end_run(run_id, output=answer)
        return answer
    except Exception as exc:
        tracer.fail_step(run_id, step_id, exc)
        tracer.fail_run(run_id, exc)
        return None

API

Method	Description
start_run(name, metadata?)	Opens a run, returns a run_id
add_step(run_id, name, step_type?, input?, metadata?)	Appends a step, returns a step_id
end_step(run_id, step_id, output?, metadata?)	Marks a step successful
fail_step(run_id, step_id, error)	Marks a step failed
end_run(run_id, output?)	Finalises with status: "success" → client.ingest
fail_run(run_id, error)	Finalises with status: "error" → client.ingest
get_run(run_id)	Returns in-progress run snapshot (no network call)