meilynx 0.3.0

Meilynx SDK for AI FinOps telemetry and outcomes ingestion.

Meilynx Python SDK

Meilynx is an AI governance and FinOps platform that gives enterprises visibility and control over LLM usage — from cost and compliance to business outcomes. This SDK lets you send structured telemetry and outcome events from your Python applications.

Install

pip install meilynx

# with OpenAI auto-instrumentation
pip install meilynx[openai]

# with Anthropic auto-instrumentation
pip install meilynx[anthropic]

# all integrations
pip install meilynx[all]

Quickstart

Option 1: Auto-instrumentation (recommended)

Add two lines at startup — your existing AI calls are tracked automatically:

from meilynx import MeilynxClient, MeilynxOptions, instrument

mx = MeilynxClient(MeilynxOptions(
    api_key="mx_live_...",  # base_url defaults to https://api.meilynx.com
))

instrument(client=mx)  # patches OpenAI, Anthropic, etc. automatically

# use your AI clients as normal — calls are tracked
from openai import OpenAI
client = OpenAI()
client.chat.completions.create(model="gpt-4o", messages=[...])

mx.shutdown()

Option 2: Drop-in import

Change one import line for explicit control:

# OpenAI
from meilynx.integrations.openai import OpenAI
client = OpenAI(meilynx_client=mx, api_key="sk-...")
client.chat.completions.create(model="gpt-4o", messages=[...])

# Anthropic
from meilynx.integrations.anthropic import Anthropic
client = Anthropic(meilynx_client=mx, api_key="sk-ant-...")
client.messages.create(model="claude-sonnet-4-20250514", max_tokens=1024, messages=[...])

Option 3: Explicit tracking

Full control over what you send:

from meilynx import MeilynxClient, MeilynxOptions
from meilynx.types import TelemetryEventInput

mx = MeilynxClient(MeilynxOptions(
    api_key="mx_live_...",  # base_url defaults to https://api.meilynx.com
))

mx.track(TelemetryEventInput(
    event_type="llm.response",
    correlation_id="run-123",
    feature_key="ask_docs",
    model="gpt-4o",
    provider="openai",
    prompt_tokens=1200,
    completion_tokens=220,
))

mx.flush()
mx.shutdown()

Which to choose? Use auto-instrumentation or drop-in imports (Options 1-2) for most cases — they automatically capture model, latency, and agentic context with zero manual work. Use explicit tracking (Option 3) when you need custom event types, non-LLM operations, or providers without built-in integration.

Context propagation with @observe

The @observe decorator propagates business context (correlation IDs, feature keys, customer IDs) through the call stack. All instrumented AI calls inside inherit this context automatically:

from meilynx import observe

@observe(feature_key="ask_docs", customer_id="acme")
def summarize(doc: str) -> str:
    # all AI calls here are tagged with feature_key="ask_docs"
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Summarize: {doc}"}],
    )
    return response.choices[0].message.content

Nested decorators inherit the parent context and can override specific fields.

Configuration

Environment variables (convention)

The SDK does not read environment variables directly. These are recommended names for your app configuration:

• MX_BASE_URL (optional) — Meilynx API URL
• MX_API_KEY — Project-scoped API key (mx_live_...)

Constructor options

Option	Type	Default	Notes
api_key	str	—	Required. API key for /v1/ingest/*.
base_url	str	"https://api.meilynx.com"	Base URL for the Meilynx API.
source_system	str	"sdk"	Source identifier.
flush_at	int	25	Batch size before flush.
flush_interval_ms	int	5000	Auto-flush interval.
max_retries	int	3	Retry attempts on 429/5xx.
retry_delay_ms	int	250	Base delay for backoff.
disable_validation	bool	False	Disable JSON schema validation.

Agentic context

For agentic loops with multiple tool-call hops, set agent_name, tool_name, and step_index on each step via @observe:

for i, step in enumerate(steps):
    @observe(agent_name="research-agent", tool_name=step.name, step_index=i)
    def run_step():
        return client.chat.completions.create(model="gpt-4o", messages=[...])
    run_step()

The auto-instrumentation also captures response_tool_calls (tool names the model invoked) automatically from streaming and non-streaming responses.

Capturing outcomes

Outcomes are the business results your AI features produce:

from meilynx import mint_idempotency_key
from meilynx.types import OutcomeEventInput

mx.capture_outcome(OutcomeEventInput(
    outcome_type="feature.result.accepted",
    idempotency_key=mint_idempotency_key("accepted", correlation_id),
    correlation_id=correlation_id,
    customer_id="cust-acme",
    feature_key="ask_docs",
))

Idempotency keys

Every outcome requires an idempotency_key to prevent duplicate processing. Use mint_idempotency_key() to generate a deterministic SHA-256 key from one or more fields:

from meilynx import mint_idempotency_key

# Same inputs always produce the same key
mint_idempotency_key("accepted", "run-123")           # → "a1b2c3..."
mint_idempotency_key("accepted", "run-123")           # → "a1b2c3..." (same)
mint_idempotency_key("accepted", "run-456")           # → "d4e5f6..." (different)

Budget status

Check current budget utilization from your application. Results are cached for 60 seconds per query-parameter combination.

status = mx.get_budget_status(customer_id="acme")

for budget in status["budgets"]:
    if budget["action"] == "block":
        print(f"Budget {budget['name']} exceeded: {budget['utilization_pct']}%")

Failsafe behavior

The SDK is designed to never break your application. All instrumentation, context injection, telemetry emission, and governance extraction are wrapped in defensive error handling:

• If context building or injection fails, the original LLM call proceeds unmodified.
• If telemetry emission fails, the error is swallowed silently.
• If governance response parsing fails, the result is returned as-is.
• Real LLM provider errors (rate limits, auth failures, invalid requests) always propagate normally.

In other words: a bug in the Meilynx SDK will log a warning but never cause your AI calls to fail.

Streaming

Auto-instrumentation handles streaming transparently. For both OpenAI and Anthropic:

• One telemetry event is emitted per completion (not per chunk)
• is_streaming is set to True automatically
• latency_ms measures time from request start to last token received
• Tool calls in the response are accumulated into response_tool_calls

No additional configuration is needed for streaming calls.

Error handling

• 429 and 5xx responses are retried with exponential backoff.
• 401/403 errors raise PermissionError immediately.

Serverless and short-lived processes

In short-lived environments (AWS Lambda, Google Cloud Functions), flush before the handler returns to avoid losing events:

# AWS Lambda
def handler(event, context):
    result = handle_request(event)
    mx.flush()          # flush before returning
    return result

# Always call shutdown() when the process is exiting
mx.shutdown()

Compatibility

• Python 3.9+
• Thread-safe batching with background flush.
• Not intended for browsers or client-side use. API keys must stay server-side.

Docs

• Documentation: docs.meilynx.com
• JS/TS SDK: github.com/meilynx/meilynx-js
• .NET SDK: github.com/meilynx/meilynx-dotnet

License

MIT