maple-python-style

Python OpenTelemetry style for Maple: module-scope tracers/meters, decorators for bounded work, error spans, OTLP-bridged logs via LoggingHandler + LoggingInstrumentor, inline endpoint + ingest key, and no helper-API wrappers.

Quality

84%

Does it follow best practices?

Run evals on this skill

Adds up to 20 points to the overall score

View guide

Securityby

Passed

No findings from the security scan

Maple Python style

pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-http opentelemetry-instrumentation-logging

Acquire OTel objects at module scope.

from opentelemetry import metrics, trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer("orders.api")
meter = metrics.get_meter("orders.api")

orders_submitted = meter.create_counter("orders.submitted", unit="1")

Bounded work

Use decorators for functions with clear boundaries. The decorator works on both sync and async def functions.

@tracer.start_as_current_span("order.submit")
async def submit_order(*, tenant_id: str, order_id: str) -> None:
    span = trace.get_current_span()
    span.set_attributes({
        "tenant.id": tenant_id,
        "order.id": order_id,
    })

Use a context manager when a decorator does not fit.

with tracer.start_as_current_span("order.validate") as span:
    span.set_attribute("tenant.id", tenant_id)
    validate_order(order)

Do not use detached tracer.start_span(...); span.end() for bounded work.

Error paths

Record exceptions on the active span.

try:
    result = await client.messages.create(...)
except Exception as exc:
    span = trace.get_current_span()
    span.record_exception(exc)
    span.set_status(Status(StatusCode.ERROR))
    logger.exception("llm call failed", extra={"tenant_id": tenant_id})
    raise

Logs

If logs are claimed as OTLP-forwarded, configure all of:

An OTel LoggerProvider + OTLPLogExporter + LoggingHandler
set_logger_provider(logger_provider) from opentelemetry._logs
Log correlation for existing records, e.g. LoggingInstrumentor().instrument(set_logging_format=True)

Preserve existing logging.basicConfig, console / file handlers, and log levels. The user's logger keeps working. The OTLP handler sits underneath so log lines carry trace_id / span_id and reach Maple.

Init behavior

Inline the endpoint and ingest key in the init module. Do not read them from env. The ingest key is project-scoped and write-only (shaped like a Sentry DSN), so source-level configuration is the default. Env indirection adds "OTel didn't start because env wasn't set" deploy failures.

# telemetry.py
import logging
import os

from opentelemetry import _logs, metrics, trace
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.logging import LoggingInstrumentor
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

MAPLE_ENDPOINT = "https://ingest.maple.dev"  # EU: https://ingest.eu.maple.dev
MAPLE_KEY = "MAPLE_TEST"  # public ingest key (maple_pk_…), or MAPLE_TEST until the user has one

_INITIALIZED = False


def init_observability() -> None:
    global _INITIALIZED
    if _INITIALIZED:
        return
    _INITIALIZED = True

    headers = {"authorization": f"Bearer {MAPLE_KEY}"}
    resource = Resource.create({
        "service.name": "my-python-app",
        "deployment.environment.name": os.getenv("DEPLOYMENT_ENV", "development"),
        "vcs.repository.url.full": "https://github.com/acme/my-python-app",
        "vcs.ref.head.revision": os.getenv("RAILWAY_GIT_COMMIT_SHA")
            or os.getenv("GITHUB_SHA")
            or os.getenv("GIT_COMMIT"),
    })

    tracer_provider = TracerProvider(resource=resource)
    tracer_provider.add_span_processor(
        BatchSpanProcessor(
            OTLPSpanExporter(endpoint=f"{MAPLE_ENDPOINT}/v1/traces", headers=headers),
        ),
    )
    trace.set_tracer_provider(tracer_provider)

    logger_provider = LoggerProvider(resource=resource)
    logger_provider.add_log_record_processor(
        BatchLogRecordProcessor(
            OTLPLogExporter(endpoint=f"{MAPLE_ENDPOINT}/v1/logs", headers=headers),
        ),
    )
    _logs.set_logger_provider(logger_provider)
    logging.getLogger().addHandler(LoggingHandler(logger_provider=logger_provider))
    LoggingInstrumentor().instrument(set_logging_format=True)

    meter_provider = MeterProvider(
        resource=resource,
        metric_readers=[
            PeriodicExportingMetricReader(
                OTLPMetricExporter(
                    endpoint=f"{MAPLE_ENDPOINT}/v1/metrics", headers=headers,
                ),
            ),
        ],
    )
    metrics.set_meter_provider(meter_provider)

Add the _INITIALIZED guard only when the app can realistically call this function more than once (FastAPI lifespan + workers, pytest fixtures, etc.).

Metrics

Counters:

llm.tokens.input
llm.tokens.output
requests/events/jobs/errors

Use UCUM units: token counters use unit="{token}". Do not add app-side llm.cost_usd pricing metrics. Maple does not price tokens; it shows cost only from gen_ai.usage.cost on the LLM span (see maple-onboarding-style "LLM calls").

Histograms:

duration
latency
payload size

Avoid raw high-cardinality values in metric attributes. Prefer tenant/org/project, operation/use case, provider/model, and outcome dimensions over user-level metric tags.

FastAPI

Use the native instrumentation (pip install opentelemetry-instrumentation-fastapi). Do not replace request handling with manual middleware.

from fastapi import FastAPI
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

import telemetry

telemetry.init_observability()
app = FastAPI()
FastAPIInstrumentor.instrument_app(app)

Import telemetry (and call init_observability()) before any other module that needs tracing.

Repository: MapleTechLabs/maple
Path: skills/maple-python-style/SKILL.md
Commit: 1a03fa7

Last updated: 11 hours ago
First committed: 5 months ago

Is this your skill?

If you maintain this skill, you can claim it as your own. Once claimed, you can manage eval scenarios, bundle related skills, attach documentation or rules, and ensure cross-agent compatibility.