Help us improve
Share bugs, ideas, or general feedback.
From open-python-skills
Structured observability with Pydantic Logfire and OpenTelemetry. Use when: (1) Adding traces/logs to Python APIs, (2) Instrumenting FastAPI, HTTPX, SQLAlchemy, or LLMs, (3) Setting up service metadata, (4) Configuring sampling or scrubbing sensitive data, (5) Testing observability code.
npx claudepluginhub jiatastic/open-python-skills --plugin open-python-skillsHow this skill is triggered — by the user, by Claude, or both
Slash command
/open-python-skills:logfireThe summary Claude sees in its skill listing — used to decide when to auto-load this skill
Structured observability for Python using Pydantic Logfire - fast setup, powerful features, OpenTelemetry-compatible.
Instruments apps with Logfire observability for traces, logs, metrics using Python, JavaScript/TypeScript, Rust SDKs and OpenTelemetry. Handles FastAPI, Django, Express, LLM calls correctly.
Patterns for structured logging, metrics collection, and distributed tracing in Python. Use when debugging production systems or adding observability.
Observability discipline: structured logging, metrics instrumentation, distributed tracing, and signal correlation. Invoke whenever task involves any interaction with observability concerns — adding logging, designing metrics, instrumenting traces, correlating signals, reviewing instrumentation, or understanding when to use which pillar.
Share bugs, ideas, or general feedback.
Structured observability for Python using Pydantic Logfire - fast setup, powerful features, OpenTelemetry-compatible.
uv pip install logfire
import logfire
logfire.configure(service_name="my-api", service_version="1.0.0")
logfire.info("Application started")
Always set service metadata at startup:
import logfire
logfire.configure(
service_name="backend",
service_version="1.0.0",
environment="production",
console=False, # Disable console output in production
send_to_logfire=True, # Send to Logfire platform
)
Instrument frameworks before creating clients/apps:
import logfire
from fastapi import FastAPI
# Configure FIRST
logfire.configure(service_name="backend")
# Then instrument
logfire.instrument_fastapi()
logfire.instrument_httpx()
logfire.instrument_sqlalchemy()
# Then create app
app = FastAPI()
# All log levels (trace → fatal)
logfire.trace("Detailed trace", step=1)
logfire.debug("Debug context", variable=locals())
logfire.info("User action", action="login", success=True)
logfire.notice("Important event", event_type="milestone")
logfire.warn("Potential issue", threshold_exceeded=True)
logfire.error("Operation failed", error_code=500)
logfire.fatal("Critical failure", component="database")
# Python 3.11+ f-string magic (auto-extracts variables)
user_id = 123
status = "active"
logfire.info(f"User {user_id} status: {status}")
# Equivalent to: logfire.info("User {user_id}...", user_id=user_id, status=status)
# Exception logging with automatic traceback
try:
risky_operation()
except Exception:
logfire.exception("Operation failed", context="extra_info")
# Spans for tracing operations
with logfire.span("Process order {order_id}", order_id="ORD-123"):
logfire.info("Validating cart")
# ... processing logic
logfire.info("Order complete")
# Dynamic span attributes
with logfire.span("Database query") as span:
results = execute_query()
span.set_attribute("result_count", len(results))
span.message = f"Query returned {len(results)} results"
# Counter - monotonically increasing
request_counter = logfire.metric_counter("http.requests", unit="1")
request_counter.add(1, {"endpoint": "/api/users", "method": "GET"})
# Gauge - current value
temperature = logfire.metric_gauge("temperature", unit="°C")
temperature.set(23.5)
# Histogram - distribution of values
latency = logfire.metric_histogram("request.duration", unit="ms")
latency.record(45.2, {"endpoint": "/api/data"})
import logfire
from pydantic_ai import Agent
logfire.configure()
logfire.instrument_pydantic_ai() # Traces all agent interactions
agent = Agent("openai:gpt-4o", system_prompt="You are helpful.")
result = agent.run_sync("Hello!")
# Suppress entire scope (e.g., noisy library)
logfire.suppress_scopes("google.cloud.bigquery.opentelemetry_tracing")
# Suppress specific code block
with logfire.suppress_instrumentation():
client.get("https://internal-healthcheck.local") # Not traced
import logfire
# Add custom patterns to scrub
logfire.configure(
scrubbing=logfire.ScrubbingOptions(
extra_patterns=["api_key", "secret", "token"]
)
)
# Custom callback for fine-grained control
def scrubbing_callback(match: logfire.ScrubMatch):
if match.path == ("attributes", "safe_field"):
return match.value # Don't scrub this field
return None # Use default scrubbing
logfire.configure(
scrubbing=logfire.ScrubbingOptions(callback=scrubbing_callback)
)
import logfire
# Sample 50% of traces
logfire.configure(sampling=logfire.SamplingOptions(head=0.5))
# Disable metrics to reduce volume
logfire.configure(metrics=False)
import logfire
from logfire.testing import CaptureLogfire
def test_user_creation(capfire: CaptureLogfire):
create_user("Alice", "alice@example.com")
spans = capfire.exporter.exported_spans
assert len(spans) >= 1
assert spans[0].attributes["user_name"] == "Alice"
capfire.exporter.clear() # Clean up for next test
| Category | Integration | Method |
|---|---|---|
| Web | FastAPI | logfire.instrument_fastapi(app) |
| Starlette | logfire.instrument_starlette(app) | |
| Django | logfire.instrument_django() | |
| Flask | logfire.instrument_flask(app) | |
| AIOHTTP Server | logfire.instrument_aiohttp_server() | |
| ASGI | logfire.instrument_asgi(app) | |
| WSGI | logfire.instrument_wsgi(app) | |
| HTTP | HTTPX | logfire.instrument_httpx() |
| Requests | logfire.instrument_requests() | |
| AIOHTTP Client | logfire.instrument_aiohttp_client() | |
| Database | SQLAlchemy | logfire.instrument_sqlalchemy(engine) |
| Asyncpg | logfire.instrument_asyncpg() | |
| Psycopg | logfire.instrument_psycopg() | |
| Redis | logfire.instrument_redis() | |
| PyMongo | logfire.instrument_pymongo() | |
| LLM | Pydantic AI | logfire.instrument_pydantic_ai() |
| OpenAI | logfire.instrument_openai() | |
| Anthropic | logfire.instrument_anthropic() | |
| MCP | logfire.instrument_mcp() | |
| Tasks | Celery | logfire.instrument_celery() |
| AWS Lambda | logfire.instrument_aws_lambda() | |
| Logging | Standard logging | logfire.instrument_logging() |
| Structlog | logfire.instrument_structlog() | |
| Loguru | logfire.instrument_loguru() | |
logfire.instrument_print() | ||
| Other | Pydantic | logfire.instrument_pydantic() |
| System Metrics | logfire.instrument_system_metrics() |
| Issue | Symptom | Fix |
|---|---|---|
| Missing service name | Spans hard to find in UI | Set service_name in configure() |
| Late instrumentation | No spans captured | Call configure() before creating clients |
| High-cardinality attrs | Storage explosion | Use IDs, not full payloads as attributes |
| Console noise | Logs pollute stdout | Set console=False in production |
configure() parameters