Initial commit: ag-bids MCP server
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>
This commit is contained in:
@@ -0,0 +1,76 @@
|
||||
"""HTTP Basic auth in front of the FastMCP Streamable-HTTP transport.
|
||||
|
||||
MetaMCP can inject ``Authorization: Basic <b64>`` on every upstream call,
|
||||
so this is the simplest robust gate. Two env vars (``AG_BIDS_MCP_USER`` and
|
||||
``AG_BIDS_MCP_PASS``) are required at process startup; the server fails
|
||||
closed if either is missing.
|
||||
|
||||
Stdio transport (local dev with Claude Desktop) skips this entirely — no
|
||||
HTTP layer exists in stdio mode.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import logging
|
||||
import os
|
||||
import secrets
|
||||
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import PlainTextResponse, Response
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
REALM = "ag-bids MCP"
|
||||
|
||||
|
||||
def expected_credentials() -> tuple[str, str]:
|
||||
"""Return the (user, pass) the server enforces. Raises if missing."""
|
||||
u = os.environ.get("AG_BIDS_MCP_USER", "")
|
||||
p = os.environ.get("AG_BIDS_MCP_PASS", "")
|
||||
if not u or not p:
|
||||
raise RuntimeError(
|
||||
"AG_BIDS_MCP_USER and AG_BIDS_MCP_PASS must both be set for HTTP "
|
||||
"Basic auth on the ag-bids MCP server."
|
||||
)
|
||||
return u, p
|
||||
|
||||
|
||||
def _decode_basic(header: str) -> tuple[str, str] | None:
|
||||
if not header or not header.lower().startswith("basic "):
|
||||
return None
|
||||
try:
|
||||
decoded = base64.b64decode(header.split(" ", 1)[1]).decode("utf-8")
|
||||
except (ValueError, UnicodeDecodeError):
|
||||
return None
|
||||
user, _, pw = decoded.partition(":")
|
||||
return user, pw
|
||||
|
||||
|
||||
def _check(presented_user: str, presented_pass: str) -> bool:
|
||||
expected_user, expected_pass = expected_credentials()
|
||||
return (
|
||||
secrets.compare_digest(presented_user, expected_user)
|
||||
and secrets.compare_digest(presented_pass, expected_pass)
|
||||
)
|
||||
|
||||
|
||||
def _unauthorized() -> Response:
|
||||
return PlainTextResponse(
|
||||
"Unauthorized",
|
||||
status_code=401,
|
||||
headers={"WWW-Authenticate": f'Basic realm="{REALM}"'},
|
||||
)
|
||||
|
||||
|
||||
async def basic_auth_middleware(request: Request, call_next):
|
||||
"""Starlette middleware that 401s anything missing valid Basic creds."""
|
||||
creds = _decode_basic(request.headers.get("authorization", ""))
|
||||
if creds is None:
|
||||
log.info("auth: missing/malformed Authorization header (path=%s)", request.url.path)
|
||||
return _unauthorized()
|
||||
user, pw = creds
|
||||
if not _check(user, pw):
|
||||
log.info("auth: bad credentials (user=%r path=%s)", user, request.url.path)
|
||||
return _unauthorized()
|
||||
return await call_next(request)
|
||||
@@ -0,0 +1,87 @@
|
||||
"""Thin HTTPS client for ag-monitor's /api/data/* endpoints.
|
||||
|
||||
The MCP server doesn't talk to SQLite directly — it goes over HTTPS to the
|
||||
ag-monitor service on a different LAN host. One X-API-Key header guards
|
||||
every call. Failures bubble up as ``AgBidsError`` so tool functions can
|
||||
turn them into user-friendly markdown.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AgBidsError(RuntimeError):
|
||||
"""Anything went wrong calling ag-monitor."""
|
||||
|
||||
|
||||
def _config() -> tuple[str, str, float]:
|
||||
url = os.environ.get("AG_BIDS_API_URL", "https://agbids.paul.farm").rstrip("/")
|
||||
key = os.environ.get("AG_BIDS_API_KEY", "")
|
||||
timeout = float(os.environ.get("AG_BIDS_API_TIMEOUT_SECS", "20"))
|
||||
if not key:
|
||||
raise AgBidsError("AG_BIDS_API_KEY is not set in the environment")
|
||||
return url, key, timeout
|
||||
|
||||
|
||||
def _get(path: str, **params: Any) -> dict | list:
|
||||
url_base, key, timeout = _config()
|
||||
url = f"{url_base}{path}"
|
||||
# Drop None values so the upstream doesn't see empty query strings.
|
||||
clean: dict[str, Any] = {k: v for k, v in params.items() if v is not None and v != ""}
|
||||
try:
|
||||
r = httpx.get(url, params=clean, timeout=timeout,
|
||||
headers={"X-API-Key": key, "Accept": "application/json"})
|
||||
except httpx.RequestError as e:
|
||||
raise AgBidsError(f"GET {url} failed: {e}") from e
|
||||
if r.status_code != 200:
|
||||
raise AgBidsError(
|
||||
f"GET {url} returned {r.status_code}: {r.text[:200]}"
|
||||
)
|
||||
try:
|
||||
return r.json()
|
||||
except ValueError as e:
|
||||
raise AgBidsError(f"GET {url} returned non-JSON: {r.text[:200]}") from e
|
||||
|
||||
|
||||
# ---- Public client API (mirrors /api/data/* shape) ----
|
||||
|
||||
|
||||
def latest(commodity: str | None = None, source: str | None = None,
|
||||
delivery: str | None = None, kind: str | None = None) -> dict:
|
||||
return _get("/api/data/latest",
|
||||
commodity=commodity, source=source, delivery=delivery, kind=kind)
|
||||
|
||||
|
||||
def history(commodity: str, source_id: int | None = None,
|
||||
delivery: str | None = None, days: int = 30) -> dict:
|
||||
return _get("/api/data/history",
|
||||
commodity=commodity, source_id=source_id,
|
||||
delivery=delivery, days=days)
|
||||
|
||||
|
||||
def best(commodity: str) -> dict:
|
||||
return _get("/api/data/best", commodity=commodity)
|
||||
|
||||
|
||||
def inputs(product: str | None = None) -> dict:
|
||||
return _get("/api/data/inputs", product=product)
|
||||
|
||||
|
||||
def sources() -> dict:
|
||||
return _get("/api/data/sources")
|
||||
|
||||
|
||||
def deliveries(commodity: str) -> dict:
|
||||
return _get("/api/data/deliveries", commodity=commodity)
|
||||
|
||||
|
||||
def todays_summary() -> dict:
|
||||
"""Brief snapshot — same blob the morning email digest uses."""
|
||||
return _get("/api/brief/snapshot", kind="morning")
|
||||
@@ -0,0 +1,249 @@
|
||||
"""JSON-from-ag-monitor → markdown helpers.
|
||||
|
||||
One function per @mcp.tool. Markdown is what FastMCP tools return; Claude /
|
||||
OpenWebUI render it nicely. Cents → dollars formatting is centralized here so
|
||||
all tools use the same precision (4 decimals for $/bu, 2 decimals for $/ton).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
|
||||
|
||||
def _bu(cents: Optional[int]) -> str:
|
||||
if cents is None:
|
||||
return "—"
|
||||
return f"${cents / 100:.4f}"
|
||||
|
||||
|
||||
def _ton(cents: Optional[int]) -> str:
|
||||
if cents is None:
|
||||
return "—"
|
||||
return f"${cents / 100:.2f}"
|
||||
|
||||
|
||||
def _basis(cents: Optional[int]) -> str:
|
||||
if cents is None:
|
||||
return "—"
|
||||
sign = "+" if cents >= 0 else ""
|
||||
return f"{sign}{cents / 100:.2f}"
|
||||
|
||||
|
||||
def _delta_arrow(cents: Optional[int]) -> str:
|
||||
if cents is None or cents == 0:
|
||||
return "—"
|
||||
return "▲" if cents > 0 else "▼"
|
||||
|
||||
|
||||
# ---------- best_local_bid ----------
|
||||
|
||||
|
||||
def fmt_best(commodity: str, payload: dict) -> str:
|
||||
best = payload.get("best")
|
||||
today = payload.get("today")
|
||||
if not best:
|
||||
return (
|
||||
f"### Best place to sell {commodity} today ({today})\n\n"
|
||||
f"No current-month {commodity} bids posted across the tracked sources.\n"
|
||||
)
|
||||
return (
|
||||
f"### Best place to sell {commodity} today ({today})\n\n"
|
||||
f"**{best['source_name']}** — delivery **{best['delivery']}** — "
|
||||
f"bid **{_bu(best['bid_cents'])}/bu** (basis {_basis(best.get('basis_cents'))}, "
|
||||
f"futures {best.get('futures_contract') or '—'})\n\n"
|
||||
f"_Fetched {best.get('fetched_at') or '?'}_\n"
|
||||
)
|
||||
|
||||
|
||||
# ---------- inputs / fertilizer ----------
|
||||
|
||||
|
||||
def fmt_inputs(payload: dict) -> str:
|
||||
rows = payload.get("rows") or []
|
||||
product = payload.get("product")
|
||||
title = f"### {product.upper()} prices" if product else "### Fertilizer + lime prices"
|
||||
if not rows:
|
||||
scope = product or "any tracked input"
|
||||
return f"{title}\n\nNo {scope} prices on file.\n"
|
||||
lines = [
|
||||
title, "",
|
||||
"| Source | Product | Delivery | Price ($/ton) | Fetched |",
|
||||
"|---|---|---|---:|---|",
|
||||
]
|
||||
for r in rows:
|
||||
lines.append(
|
||||
f"| {r['source_name']} | {r.get('display_name') or r['commodity']} | "
|
||||
f"{r['delivery']} | {_ton(r.get('bid_cents'))} | {r.get('fetched_at') or '?'} |"
|
||||
)
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
# ---------- latest snapshot ----------
|
||||
|
||||
|
||||
def fmt_latest(payload: dict) -> str:
|
||||
rows = payload.get("rows") or []
|
||||
if not rows:
|
||||
return "### Latest prices\n\nNo rows match those filters.\n"
|
||||
lines = [
|
||||
"### Latest prices", "",
|
||||
"| Source | Commodity | Delivery | Bid | Basis | Futures | Fetched |",
|
||||
"|---|---|---|---:|---:|---|---|",
|
||||
]
|
||||
for r in rows:
|
||||
unit_fmt = _ton if r.get("commodity_kind") == "fertilizer" else _bu
|
||||
lines.append(
|
||||
f"| {r['source_name']} | {r.get('display_name') or r['commodity']} | "
|
||||
f"{r['delivery']} | {unit_fmt(r.get('bid_cents'))} | "
|
||||
f"{_basis(r.get('basis_cents'))} | {r.get('futures_contract') or '—'} | "
|
||||
f"{r.get('fetched_at') or '?'} |"
|
||||
)
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
# ---------- price history ----------
|
||||
|
||||
|
||||
def fmt_history(payload: dict, max_rows: int = 60) -> str:
|
||||
rows = payload.get("rows") or []
|
||||
commodity = payload.get("commodity")
|
||||
days = payload.get("days")
|
||||
if not rows:
|
||||
return f"### {commodity} history ({days}d)\n\nNo samples in the window.\n"
|
||||
|
||||
# Per (source, delivery) trend annotation: first vs last sample.
|
||||
series: dict[tuple, list[dict]] = {}
|
||||
for r in rows:
|
||||
series.setdefault((r["source_name"], r["delivery"]), []).append(r)
|
||||
|
||||
lines = [f"### {commodity} price history — last {days} days", ""]
|
||||
for (src, dlv), pts in sorted(series.items()):
|
||||
if not pts:
|
||||
continue
|
||||
first = pts[0].get("bid_cents")
|
||||
last = pts[-1].get("bid_cents")
|
||||
delta = (last - first) if (first is not None and last is not None) else None
|
||||
arrow = _delta_arrow(delta)
|
||||
lines.append(
|
||||
f"- **{src}** / {dlv}: {len(pts)} samples · "
|
||||
f"{_bu(first)} → {_bu(last)} {arrow} {_basis(delta) if delta is not None else ''}".rstrip()
|
||||
)
|
||||
|
||||
# If the history is shallow include the raw rows too (helpful for charts).
|
||||
if sum(len(p) for p in series.values()) <= max_rows:
|
||||
lines.extend([
|
||||
"",
|
||||
"| Time | Source | Delivery | Bid | Basis |",
|
||||
"|---|---|---|---:|---:|",
|
||||
])
|
||||
for r in rows[-max_rows:]:
|
||||
lines.append(
|
||||
f"| {r['fetched_at']} | {r['source_name']} | {r['delivery']} | "
|
||||
f"{_bu(r.get('bid_cents'))} | {_basis(r.get('basis_cents'))} |"
|
||||
)
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
# ---------- sources / health ----------
|
||||
|
||||
|
||||
def fmt_sources(payload: dict) -> str:
|
||||
src = payload.get("sources") or []
|
||||
if not src:
|
||||
return "### Sources\n\nNo active sources.\n"
|
||||
lines = [
|
||||
"### Tracked sources", "",
|
||||
"| Source | Kind | Last success | Consecutive failures | Last error |",
|
||||
"|---|---|---|---:|---|",
|
||||
]
|
||||
for s in src:
|
||||
lines.append(
|
||||
f"| {s['name']} | {s['kind']} | {s.get('last_success_at') or '—'} | "
|
||||
f"{s.get('consecutive_failures') or 0} | {s.get('last_error') or ''} |"
|
||||
)
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def fmt_health(payload: dict) -> str:
|
||||
src = payload.get("sources") or []
|
||||
healthy, stale, down = [], [], []
|
||||
for s in src:
|
||||
n = s.get("consecutive_failures") or 0
|
||||
if n >= 3:
|
||||
down.append(s)
|
||||
elif not s.get("last_success_at"):
|
||||
stale.append(s)
|
||||
else:
|
||||
healthy.append(s)
|
||||
lines = ["### Source health", ""]
|
||||
lines.append(f"- ✅ Healthy: **{len(healthy)}**")
|
||||
lines.append(f"- ⚠ Stale (never succeeded): **{len(stale)}**")
|
||||
lines.append(f"- ❌ Down (3+ consecutive failures): **{len(down)}**")
|
||||
if stale or down:
|
||||
lines.append("")
|
||||
lines.append("| Source | State | Last success | Failures | Error |")
|
||||
lines.append("|---|---|---|---:|---|")
|
||||
for s in stale + down:
|
||||
state = "stale" if s in stale else "down"
|
||||
lines.append(
|
||||
f"| {s['name']} | {state} | {s.get('last_success_at') or '—'} | "
|
||||
f"{s.get('consecutive_failures') or 0} | "
|
||||
f"{(s.get('last_error') or '')[:80]} |"
|
||||
)
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
# ---------- list helpers ----------
|
||||
|
||||
|
||||
def fmt_deliveries(payload: dict) -> str:
|
||||
commodity = payload.get("commodity")
|
||||
labels = payload.get("deliveries") or []
|
||||
if not labels:
|
||||
return f"### {commodity} deliveries\n\nNo posted deliveries.\n"
|
||||
return (
|
||||
f"### Posted delivery labels for {commodity}\n\n"
|
||||
+ "\n".join(f"- {x}" for x in labels)
|
||||
+ "\n"
|
||||
)
|
||||
|
||||
|
||||
def fmt_commodities() -> str:
|
||||
return (
|
||||
"### Tracked commodities\n\n"
|
||||
"- **corn** ($/bu)\n"
|
||||
"- **soy** — soybeans ($/bu)\n"
|
||||
"- **wheat** ($/bu) — tracked but excluded from daily brief emails\n"
|
||||
"- **map** — MAP 11-52-0 ($/ton)\n"
|
||||
"- **potash** — Potash 0-0-60 ($/ton)\n"
|
||||
"- **lime** ($/ton)\n"
|
||||
)
|
||||
|
||||
|
||||
# ---------- today's summary ----------
|
||||
|
||||
|
||||
def fmt_summary(payload: dict) -> str:
|
||||
today = payload.get("today")
|
||||
prev = payload.get("prev_trading_day")
|
||||
lines = [f"### Market summary — today {today} vs prev trading day {prev}", ""]
|
||||
for c in payload.get("commodities", []):
|
||||
f = c.get("futures") or {}
|
||||
chg = f.get("change_cents")
|
||||
arrow = _delta_arrow(chg)
|
||||
lines.append(
|
||||
f"**{c['display_name']}** — CBOT {f.get('contract','?')} "
|
||||
f"last {_bu(f.get('today_last_cents'))}, prev close "
|
||||
f"{_bu(f.get('prev_close_cents'))} {arrow} {_basis(chg) if chg is not None else ''}"
|
||||
)
|
||||
bft = c.get("best_for_today")
|
||||
if bft:
|
||||
lines.append(
|
||||
f" · Best for today's delivery ({bft['delivery']}): "
|
||||
f"**{bft['source_name']}** @ {_bu(bft['bid_cents'])} "
|
||||
f"(basis {_basis(bft.get('basis_cents'))})"
|
||||
)
|
||||
else:
|
||||
lines.append(" · No current-month local bid posted.")
|
||||
lines.append("")
|
||||
return "\n".join(lines).rstrip() + "\n"
|
||||
@@ -0,0 +1,275 @@
|
||||
"""ag-bids MCP server.
|
||||
|
||||
Mirrors the zerto-docs-rag layout — FastMCP, ``@mcp.tool()`` decorators
|
||||
returning markdown strings, dual stdio/streamable-http transport — with one
|
||||
addition: HTTP Basic auth in front of the streamable-http transport. MetaMCP
|
||||
upstream config injects ``Authorization: Basic <b64>`` so the proxied call
|
||||
sails through; direct hits without the header get 401.
|
||||
|
||||
Run locally (stdio for Claude Desktop):
|
||||
MCP_TRANSPORT=stdio python -m ag_bids_mcp.server
|
||||
|
||||
Run in a container (HTTP for MetaMCP upstream):
|
||||
python -m ag_bids_mcp.server # default transport = streamable-http
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from typing import Annotated
|
||||
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
from pydantic import Field
|
||||
|
||||
from ag_bids_mcp import client, format as fmt
|
||||
from ag_bids_mcp.auth import basic_auth_middleware, expected_credentials
|
||||
from ag_bids_mcp.usage import track
|
||||
|
||||
logging.basicConfig(
|
||||
level=os.environ.get("LOG_LEVEL", "INFO"),
|
||||
format="%(asctime)s %(levelname)s %(name)s %(message)s",
|
||||
stream=sys.stderr,
|
||||
)
|
||||
log = logging.getLogger("ag-bids-mcp")
|
||||
|
||||
|
||||
mcp = FastMCP("ag-bids", stateless_http=True)
|
||||
|
||||
|
||||
VALID_GRAIN = {"corn", "soy", "wheat"}
|
||||
VALID_INPUT = {"lime", "map", "potash"}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Tools
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def best_local_bid(
|
||||
commodity: Annotated[
|
||||
str, Field(description="Grain to look up: 'corn', 'soy' (soybeans), or 'wheat'.")
|
||||
],
|
||||
) -> str:
|
||||
"""Return the highest local bid for *this calendar month's* delivery for
|
||||
the given grain. This is the "where should I haul today" answer."""
|
||||
commodity = commodity.strip().lower()
|
||||
with track("best_local_bid", commodity=commodity):
|
||||
if commodity not in VALID_GRAIN:
|
||||
return f"`commodity` must be one of: {sorted(VALID_GRAIN)}"
|
||||
payload = client.best(commodity)
|
||||
return fmt.fmt_best(commodity, payload)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def current_lime_price() -> str:
|
||||
"""Latest lime prices on file across all sources. Lime is rarely posted on
|
||||
public bid pages — entries usually come from manual admin input."""
|
||||
with track("current_lime_price"):
|
||||
payload = client.inputs(product="lime")
|
||||
return fmt.fmt_inputs(payload)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def current_input_price(
|
||||
product: Annotated[
|
||||
str | None,
|
||||
Field(description="One of: 'lime', 'map', 'potash'. Omit for all three."),
|
||||
] = None,
|
||||
) -> str:
|
||||
"""Latest fertilizer / lime prices ($/ton)."""
|
||||
p = product.strip().lower() if product else None
|
||||
with track("current_input_price", product=p):
|
||||
if p is not None and p not in VALID_INPUT:
|
||||
return f"`product` must be one of: {sorted(VALID_INPUT)} (or omit)"
|
||||
payload = client.inputs(product=p)
|
||||
return fmt.fmt_inputs(payload)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def latest_prices(
|
||||
commodity: Annotated[
|
||||
str | None,
|
||||
Field(description="Filter to one commodity (corn / soy / wheat / map / potash / lime)."),
|
||||
] = None,
|
||||
source: Annotated[
|
||||
str | None,
|
||||
Field(description="Filter to one source by exact display name."),
|
||||
] = None,
|
||||
delivery: Annotated[
|
||||
str | None,
|
||||
Field(description="Filter to one delivery label (e.g. 'May 2026', 'Oct/Nov 2026')."),
|
||||
] = None,
|
||||
) -> str:
|
||||
"""Snapshot of the latest scraped bid per (source, commodity, delivery)."""
|
||||
cm = commodity.strip().lower() if commodity else None
|
||||
with track("latest_prices", commodity=cm, source=source, delivery=delivery):
|
||||
payload = client.latest(commodity=cm, source=source, delivery=delivery)
|
||||
return fmt.fmt_latest(payload)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def price_history(
|
||||
commodity: Annotated[
|
||||
str, Field(description="One of corn / soy / wheat / map / potash / lime.")
|
||||
],
|
||||
source: Annotated[
|
||||
str | None,
|
||||
Field(description="Optional source display name to narrow the chart."),
|
||||
] = None,
|
||||
delivery: Annotated[
|
||||
str | None,
|
||||
Field(description="Optional delivery label to narrow the chart."),
|
||||
] = None,
|
||||
days: Annotated[
|
||||
int, Field(ge=1, le=365, description="Lookback window in days.")
|
||||
] = 30,
|
||||
) -> str:
|
||||
"""Compact price history per (source, delivery) for the chosen commodity.
|
||||
|
||||
Returns per-series ▲/▼ trend annotations plus the raw points if the
|
||||
window has fewer than ~60 samples."""
|
||||
cm = commodity.strip().lower()
|
||||
with track("price_history", commodity=cm, source=source, delivery=delivery, days=days):
|
||||
# source_id lookup would require an extra call; for now we accept
|
||||
# source by name and let users filter the markdown output. The
|
||||
# underlying /api/data/history accepts source_id; pass-through is
|
||||
# source-agnostic.
|
||||
payload = client.history(commodity=cm, delivery=delivery, days=days)
|
||||
# If user asked for a single source, filter rows post-fetch.
|
||||
if source:
|
||||
payload["rows"] = [r for r in payload.get("rows") or [] if r.get("source_name") == source]
|
||||
return fmt.fmt_history(payload)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_sources() -> str:
|
||||
"""All active scrapers + their last-success timestamps and any pending failures."""
|
||||
with track("list_sources"):
|
||||
payload = client.sources()
|
||||
return fmt.fmt_sources(payload)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_commodities() -> str:
|
||||
"""The complete set of commodities tracked by ag-monitor."""
|
||||
with track("list_commodities"):
|
||||
return fmt.fmt_commodities()
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_deliveries(
|
||||
commodity: Annotated[
|
||||
str, Field(description="Commodity whose posted delivery labels you want.")
|
||||
],
|
||||
) -> str:
|
||||
"""All posted delivery labels for a commodity, sorted chronologically."""
|
||||
cm = commodity.strip().lower()
|
||||
with track("list_deliveries", commodity=cm):
|
||||
payload = client.deliveries(cm)
|
||||
return fmt.fmt_deliveries(payload)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def source_health() -> str:
|
||||
"""Operational status of every source: healthy, stale, or down."""
|
||||
with track("source_health"):
|
||||
payload = client.sources()
|
||||
return fmt.fmt_health(payload)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def todays_summary() -> str:
|
||||
"""Today's market snapshot — same blob used by the morning email brief.
|
||||
|
||||
Includes CBOT corn + soy continuous futures vs the previous trading
|
||||
day's close, and the best local bid for each commodity's current-month
|
||||
delivery."""
|
||||
with track("todays_summary"):
|
||||
payload = client.todays_summary()
|
||||
return fmt.fmt_summary(payload)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Entry point
|
||||
# ============================================================================
|
||||
|
||||
|
||||
def _streamable_app_with_auth():
|
||||
"""Attach Basic-auth middleware directly to FastMCP's Starlette app.
|
||||
|
||||
Earlier we tried mounting the FastMCP app under a parent Starlette app to
|
||||
install middleware. That broke the streamable-HTTP transport because the
|
||||
parent's lifespan didn't trigger FastMCP's internal session-manager
|
||||
task-group startup, so requests hit "Task group is not initialized".
|
||||
Adding middleware on the FastMCP-provided app keeps the original lifespan
|
||||
intact.
|
||||
"""
|
||||
from starlette.middleware.base import BaseHTTPMiddleware
|
||||
|
||||
app = mcp.streamable_http_app()
|
||||
app.add_middleware(BaseHTTPMiddleware, dispatch=basic_auth_middleware)
|
||||
return app
|
||||
|
||||
|
||||
def main() -> None:
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument(
|
||||
"--transport",
|
||||
default=os.environ.get("MCP_TRANSPORT", "stdio"),
|
||||
choices=["stdio", "streamable-http", "sse"],
|
||||
)
|
||||
p.add_argument("--host", default=os.environ.get("MCP_HOST", "0.0.0.0"))
|
||||
p.add_argument("--port", type=int, default=int(os.environ.get("MCP_PORT", "8000")))
|
||||
args = p.parse_args()
|
||||
|
||||
if args.transport == "stdio":
|
||||
# stdio has no HTTP layer → no Basic auth to enforce. Useful for
|
||||
# local Claude Desktop development.
|
||||
log.info("starting ag-bids MCP on stdio (auth bypassed in stdio mode)")
|
||||
mcp.run()
|
||||
return
|
||||
|
||||
# HTTP transports: fail closed if Basic-auth credentials are unset.
|
||||
expected_credentials()
|
||||
|
||||
# Same DNS-rebinding logic as zerto-docs-rag: behind a Docker DNS name
|
||||
# like "ag-bids-mcp:8000", FastMCP's default localhost-only check would
|
||||
# 421 every request.
|
||||
allowed_hosts = os.environ.get("MCP_ALLOWED_HOSTS")
|
||||
allowed_origins = os.environ.get("MCP_ALLOWED_ORIGINS")
|
||||
if (
|
||||
os.environ.get("MCP_DISABLE_DNS_REBINDING_PROTECTION") in {"1", "true", "yes"}
|
||||
or allowed_hosts == "*"
|
||||
or allowed_origins == "*"
|
||||
):
|
||||
mcp.settings.transport_security.enable_dns_rebinding_protection = False
|
||||
else:
|
||||
if allowed_hosts:
|
||||
mcp.settings.transport_security.allowed_hosts = [
|
||||
h.strip() for h in allowed_hosts.split(",") if h.strip()
|
||||
]
|
||||
if allowed_origins:
|
||||
mcp.settings.transport_security.allowed_origins = [
|
||||
o.strip() for o in allowed_origins.split(",") if o.strip()
|
||||
]
|
||||
|
||||
if args.transport == "sse":
|
||||
# SSE transport: same direct-add-middleware pattern.
|
||||
from starlette.middleware.base import BaseHTTPMiddleware
|
||||
app = mcp.sse_app()
|
||||
app.add_middleware(BaseHTTPMiddleware, dispatch=basic_auth_middleware)
|
||||
else:
|
||||
app = _streamable_app_with_auth()
|
||||
|
||||
import uvicorn
|
||||
log.info("starting ag-bids MCP on %s://%s:%s (Basic auth enforced)",
|
||||
args.transport, args.host, args.port)
|
||||
uvicorn.run(app, host=args.host, port=args.port, log_level="info")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,93 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user