示踪剂
用于分布式跟踪和跨度管理的轻量级Python可观测性SDK。不需要外部收集器。
______________________________________________________________________
特性
- 上下文管理器跨度 --开始和结束跨度
with tracer.start_span(...)，自动捕获开始/结束时间和持续时间。 - 装饰仪器 --用以下命令包裹任何同步或异步函数
@trace/@trace_async；捕获异常，并将范围标记为ERROR。 - 可插拔存储 --船舶与
InMemoryStorage（默认）和FileStorage用于持久的、文件支持的跟踪。 - MCP服务器 --通过模型上下文协议公开实时跟踪数据
list_traces,get_trace，以及get_stats工具。 - 丰富的终端仪表板 --使用以下工具在终端中呈现跟踪和聚合统计信息
dashboard模块。 - 线程和异步安全 --通过螺纹锁保护存储；主动跨度跟踪用途
contextvars.ContextVar以实现正确的异步隔离。
______________________________________________________________________
安装
pip install traceagent # production
pip install "traceagent[dev]" # includes pytest, pytest-asyncio, ruff或来源：
git clone https://github.com/techknowmad/trace-agent.git
cd trace-agent
pip install -e ".[dev]"______________________________________________________________________
快速开始
from traceagent import get_tracer, trace, trace_async
# --- Context manager ---
tracer = get_tracer()
with tracer.start_span("db.query", attributes={"table": "users"}) as span:
span.add_event("cache_miss")
rows = fetch_rows() # your code here
# --- Sync decorator ---
@trace(name="process-request")
def handle(request):
return {"ok": True}
# --- Async decorator ---
@trace_async(name="fetch-data")
async def fetch(url: str):
async with httpx.AsyncClient() as client:
return await client.get(url)
# --- Persistent storage ---
from traceagent import FileStorage, Tracer
tracer = Tracer(storage=FileStorage("/tmp/traces"))
with tracer.start_span("batch-job") as span:
span.set_attribute("records", 1_000)______________________________________________________________________
MCP服务器
from traceagent.mcp_server import MCPServer
server = MCPServer()
# List all recorded traces
traces = server.call_tool("list_traces")
# Retrieve a specific trace by ID
trace = server.call_tool("get_trace", {"trace_id": ""})
# Aggregate statistics
stats = server.call_tool("get_stats")______________________________________________________________________
建筑
traceagent/
├── models.py # Span, Trace, SpanStatus — pure dataclasses, no I/O
├── tracer.py # Tracer — span lifecycle, ContextVar active-span tracking
├── storage.py # InMemoryStorage, FileStorage — thread-safe backends
├── decorators.py # @trace, @trace_async — wraps functions, captures errors
├── mcp_server.py # MCPServer — MCP-protocol tool surface over live storage
└── dashboard.py # Rich-powered terminal renderer for traces and stats数据流：
caller
└─ Tracer.start_span()
├─ creates Span (model)
├─ sets ContextVar (active span)
└─ on __exit__ / exception
├─ records end_time, duration_ms, SpanStatus
└─ Storage.save_span()
├─ InMemoryStorage → dict in process memory
└─ FileStorage → JSON files on disk
MCPServer
└─ reads from Storage → serves list_traces / get_trace / get_stats______________________________________________________________________
发展
# Run all tests
pytest -v
# Lint
ruff check .
# Run a single test module
pytest tests/test_tracer.py -v______________________________________________________________________
贡献
看 贡献.md 用于分支约定、代码风格和拉取请求指南。
______________________________________________________________________
许可证
麻省理工学院 ©2026 TechKnowMad实验室私人有限公司
______________________________________________________________________
建造于 TechKnowMad实验室