justin
875a190983
Exposes live + historical ag-bids commodity data (from the ag-monitor service at agbids.paul.farm) as MCP tools, sitting behind MetaMCP at https://mcp.jpaul.io/metamcp/ag-bids/mcp. Pattern mirrors zerto-docs-rag with one addition: HTTP Basic auth in front of the streamable-HTTP transport so namespace guessers can't reach the tools. Stdio transport is unaffected (used by local Claude Desktop dev). Tools (markdown returns, ~15 LOC each): best_local_bid(commodity) — where to sell corn/soy/wheat today, for the current calendar month only current_lime_price() — latest lime quotes ($/ton) current_input_price(product?) — MAP / Potash / Lime latest_prices(...) — filtered snapshot price_history(...) — per-(source,delivery) trend list_sources / list_commodities / list_deliveries source_health() — healthy / stale / down buckets todays_summary() — same shape as morning brief snapshot Data path: ag-bids-mcp -> X-API-Key -> /api/data/* on ag-monitor (reuses BRIEF_API_KEY). Tests: 24 covering the httpx client, markdown formatters, HTTP Basic middleware (401/200), and JSONL usage logging. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
94 lines
2.6 KiB
Python
94 lines
2.6 KiB
Python
"""Per-tool-call usage logging (one JSONL line per invocation).
|
|
|
|
Trimmed from zerto-docs-rag/docs_mcp/usage.py. Captures: timestamp, tool
|
|
name, args (commodity / source / delivery / etc — all non-PII), success
|
|
flag, error class on failure, elapsed ms. Useful for: spotting hot tools,
|
|
seeing which queries fail upstream, weekly summaries.
|
|
|
|
Writes to ``$USAGE_LOG_DIR/usage-YYYY-MM-DD.jsonl``, one file per day,
|
|
auto-rotated. Old files beyond ``USAGE_LOG_KEEP_DAYS`` are deleted on
|
|
each write.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def _log_dir() -> Path | None:
|
|
raw = os.environ.get("USAGE_LOG_DIR", "")
|
|
if not raw:
|
|
return None
|
|
p = Path(raw)
|
|
try:
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
except OSError as e:
|
|
log.warning("usage log dir %s unavailable: %s", raw, e)
|
|
return None
|
|
return p
|
|
|
|
|
|
def _keep_days() -> int:
|
|
try:
|
|
return int(os.environ.get("USAGE_LOG_KEEP_DAYS", "90"))
|
|
except ValueError:
|
|
return 90
|
|
|
|
|
|
def _prune(dir_: Path) -> None:
|
|
cutoff = (datetime.now(timezone.utc) - timedelta(days=_keep_days())).date()
|
|
for f in dir_.glob("usage-*.jsonl"):
|
|
try:
|
|
date_part = f.stem.removeprefix("usage-")
|
|
file_date = datetime.strptime(date_part, "%Y-%m-%d").date()
|
|
if file_date < cutoff:
|
|
f.unlink()
|
|
except (ValueError, OSError):
|
|
continue
|
|
|
|
|
|
def write(record: dict[str, Any]) -> None:
|
|
dir_ = _log_dir()
|
|
if dir_ is None:
|
|
return
|
|
today = datetime.now(timezone.utc).date().isoformat()
|
|
path = dir_ / f"usage-{today}.jsonl"
|
|
try:
|
|
with path.open("a", encoding="utf-8") as fp:
|
|
json.dump(record, fp, default=str)
|
|
fp.write("\n")
|
|
_prune(dir_)
|
|
except OSError as e:
|
|
log.warning("usage log write failed: %s", e)
|
|
|
|
|
|
@contextmanager
|
|
def track(tool: str, **fields: Any):
|
|
"""Wrap a tool body. Logs on both success and failure with elapsed ms."""
|
|
started = time.perf_counter()
|
|
record: dict[str, Any] = {
|
|
"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
|
|
"tool": tool,
|
|
**fields,
|
|
}
|
|
try:
|
|
yield record
|
|
record["ok"] = True
|
|
except Exception as e:
|
|
record["ok"] = False
|
|
record["error_class"] = type(e).__name__
|
|
record["error_msg"] = str(e)[:200]
|
|
raise
|
|
finally:
|
|
record["elapsed_ms"] = int((time.perf_counter() - started) * 1000)
|
|
write(record)
|