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>
78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
"""Usage-log smoke tests — make sure JSONL records land on disk."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
|
|
|
|
def test_track_writes_jsonl_record(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("USAGE_LOG_DIR", str(tmp_path))
|
|
monkeypatch.setenv("USAGE_LOG_KEEP_DAYS", "90")
|
|
from ag_bids_mcp import usage
|
|
importlib.reload(usage)
|
|
|
|
with usage.track("best_local_bid", commodity="corn"):
|
|
pass
|
|
|
|
files = list(tmp_path.glob("usage-*.jsonl"))
|
|
assert len(files) == 1
|
|
line = files[0].read_text().strip().splitlines()[-1]
|
|
rec = json.loads(line)
|
|
assert rec["tool"] == "best_local_bid"
|
|
assert rec["commodity"] == "corn"
|
|
assert rec["ok"] is True
|
|
assert "elapsed_ms" in rec
|
|
|
|
|
|
def test_track_records_failure(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("USAGE_LOG_DIR", str(tmp_path))
|
|
from ag_bids_mcp import usage
|
|
importlib.reload(usage)
|
|
|
|
import pytest
|
|
with pytest.raises(RuntimeError):
|
|
with usage.track("price_history", commodity="corn"):
|
|
raise RuntimeError("boom")
|
|
|
|
files = list(tmp_path.glob("usage-*.jsonl"))
|
|
rec = json.loads(files[0].read_text().splitlines()[-1])
|
|
assert rec["ok"] is False
|
|
assert rec["error_class"] == "RuntimeError"
|
|
assert "boom" in rec["error_msg"]
|
|
|
|
|
|
def test_track_no_log_dir_is_silent(monkeypatch):
|
|
monkeypatch.delenv("USAGE_LOG_DIR", raising=False)
|
|
from ag_bids_mcp import usage
|
|
importlib.reload(usage)
|
|
|
|
# Should not raise even though no log dir is set
|
|
with usage.track("noop"):
|
|
pass
|
|
|
|
|
|
def test_prune_removes_old_files(monkeypatch, tmp_path):
|
|
from datetime import datetime, timedelta, timezone
|
|
monkeypatch.setenv("USAGE_LOG_DIR", str(tmp_path))
|
|
monkeypatch.setenv("USAGE_LOG_KEEP_DAYS", "1")
|
|
from ag_bids_mcp import usage
|
|
importlib.reload(usage)
|
|
|
|
# Create a fake old log file
|
|
old_date = (datetime.now(timezone.utc) - timedelta(days=5)).date().isoformat()
|
|
old_file = tmp_path / f"usage-{old_date}.jsonl"
|
|
old_file.write_text('{"old": true}\n')
|
|
assert old_file.exists()
|
|
|
|
# Writing a new record triggers prune
|
|
with usage.track("ping"):
|
|
pass
|
|
|
|
assert not old_file.exists(), "expected old log file to be pruned"
|