megflow-observability

Python SDK for Megflow AI Observability — track tokens, cost, latency and errors across OpenAI, Anthropic, Gemini and more.

Install

pip install megflow-observability

Quick start

from megflow_observability import MegflowObserve

observe = MegflowObserve(api_key="obs_your_key_here")

observe.track(
    provider="openai",
    model="gpt-4o",
    input_tokens=100,
    output_tokens=50,
    total_tokens=150,
    cost_usd=0.00075,
    latency_ms=320,
    status_code=200,
)

Auto-instrument OpenAI

from openai import OpenAI
from megflow_observability import MegflowObserve, wrap_openai

client = wrap_openai(OpenAI(), MegflowObserve(api_key="obs_your_key_here"))

# All calls are tracked automatically
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

Auto-instrument Anthropic

import anthropic
from megflow_observability import MegflowObserve, wrap_anthropic

client = wrap_anthropic(anthropic.Anthropic(), MegflowObserve(api_key="obs_your_key_here"))

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
)

Auto-instrument Gemini

import google.generativeai as genai
from megflow_observability import MegflowObserve, wrap_gemini

genai.configure(api_key="YOUR_GOOGLE_API_KEY")
observe = MegflowObserve(api_key="obs_your_key_here")

model = wrap_gemini(genai.GenerativeModel("gemini-2.0-flash"), observe)
response = model.generate_content("Hello")

Auto-instrument AWS Bedrock

Works with Claude, Llama, Mistral, Titan and Cohere on Bedrock.

import boto3
import json
from megflow_observability import MegflowObserve, wrap_bedrock

observe = MegflowObserve(api_key="obs_your_key_here")
client = wrap_bedrock(
    boto3.client("bedrock-runtime", region_name="us-east-1"),
    observe,
)

response = client.invoke_model(
    modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
    body=json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "messages": [{"role": "user", "content": "Hello"}],
        "max_tokens": 1024,
    }),
)
result = json.loads(response["body"].read())

Auto-instrument Cohere

Works with both Client (v1) and ClientV2 (v2).

import cohere
from megflow_observability import MegflowObserve, wrap_cohere

observe = MegflowObserve(api_key="obs_your_key_here")

# v2 API
co = wrap_cohere(cohere.ClientV2("COHERE_API_KEY"), observe)
response = co.chat(
    model="command-r-plus",
    messages=[{"role": "user", "content": "Hello"}],
)

# v1 API
co = wrap_cohere(cohere.Client("COHERE_API_KEY"), observe)
response = co.chat(model="command-r-plus", message="Hello")

Prompt observability

Every wrap_*() function auto-computes a prompt_size (character count) and prompt_hash (SHA-256, first 16 chars) from the prompt-relevant input before sending anything — the raw prompt text itself is never transmitted. This lets the dashboard's Prompts page detect and group repeated prompts.

prompt_version has no natural source in an API call, so tag it explicitly per wrapped client:

client = wrap_openai(OpenAI(), observe, prompt_version="v3")

Optional dependencies

pip install megflow-observability[openai]      # includes openai
pip install megflow-observability[anthropic]   # includes anthropic
pip install megflow-observability[gemini]      # includes google-generativeai
pip install megflow-observability[all]         # includes all three

Links

• Dashboard: observability.megflow.com
• JS SDK: npmjs.com/package/megflow-observability