From e5b42287ecbba3aa16891670391a27e3493b7e16 Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Sat, 18 Jul 2026 05:08:41 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 730 +++++++++++-------------------------------------------- 1 file changed, 140 insertions(+), 590 deletions(-) diff --git a/agent.py b/agent.py index 8485140..ea75de1 100644 --- a/agent.py +++ b/agent.py @@ -1,639 +1,189 @@ -"""QuoteJudge Studio full-stack A2A agent. +"""quote-judge-studio-v1 agent. -Deterministic, no-LLM quote comparison product: -- platform-authenticated user tenancy; -- managed Postgres persistence; -- typed A2A/MCP tools for comparison, reload, browser uploads, API uploads; -- production receipt records without secrets. +Starter stack: + - DeepAgents for tool-calling orchestration + - Caller-provided LLM credentials via ctx.llm + - A tiny model-call middleware hook you can replace with tracing, + routing, rate limits, or policy checks """ from __future__ import annotations -import base64 -import csv -import hashlib -import io import json -import os -import re -from datetime import datetime, timezone -from typing import Annotated, Any +from pathlib import Path +from typing import Any -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel import a2a_pack as a2a from a2a_pack import ( A2AAgent, - AgentDatabase, - AgentDatabaseEnv, - AgentDatabaseMigrations, - AgentPlatformResources, - FileUpload, - PlatformUserAuth, + LLMProvisioning, + {{ auth_type }}, Pricing, - Resources, RunContext, - UploadedFile, WorkspaceAccess, WorkspaceMode, ) - -MAX_QUOTES = 25 -MAX_BROWSER_DOCUMENTS = 4 -MAX_BROWSER_DOCUMENT_BYTES = 64 * 1024 -ACCEPTED_UPLOAD_TYPES = ( - "application/json", - "text/csv", - "text/plain", -) -DATABASE_ENV = "DATABASE_URL" +from a2a_pack.context import LLMCreds class QuoteJudgeStudioV1Config(BaseModel): - """No user-editable config is required.""" + pass -class QuoteInput(BaseModel): - vendor: Annotated[str, Field(min_length=1, max_length=160)] - unit_price: Annotated[float, Field(gt=0, le=1_000_000)] - quantity: Annotated[int, Field(gt=0, le=10_000_000)] - delivery_days: Annotated[int, Field(ge=0, le=3650)] - warranty_months: Annotated[int, Field(ge=0, le=600)] +SYSTEM_PROMPT = """\ +You are a compact tool-calling agent. - @field_validator("vendor") - @classmethod - def _clean_vendor(cls, value: str) -> str: - cleaned = re.sub(r"\s+", " ", value).strip() - if not cleaned: - raise ValueError("vendor is required") - return cleaned +Use the text_stats tool when the user asks about text, counts, summaries, +or anything where exact length/word numbers would help. Mention tool results +briefly instead of dumping raw JSON. +""" + +RUNTIME_SKILLS_DIR = "quote-judge-studio-v1/.deepagents/skills/" +DEEPAGENTS_RECURSION_LIMIT = 500 -class QuoteWeights(BaseModel): - price: Annotated[float, Field(ge=0, le=100)] = 50 - delivery: Annotated[float, Field(ge=0, le=100)] = 30 - warranty: Annotated[float, Field(ge=0, le=100)] = 20 - - -class BrowserDocument(BaseModel): - filename: Annotated[str, Field(min_length=1, max_length=180)] - media_type: Annotated[str, Field(min_length=1, max_length=100)] - data_base64: Annotated[str, Field(min_length=1, max_length=90_000)] - - -class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]): +class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, {{ auth_type }}]): name = "quote-judge-studio-v1" - description = ( - "QuoteJudge compares vendor quotes with weighted price, delivery, " - "and warranty scoring, persists recommendations, supports upload/paste " - "workflows, and exposes typed A2A/MCP tools." - ) - version = "0.1.2" + description = "One-page QuoteJudge studio for comparing vendor quotes with weighted price, delivery, and warranty scoring, persistent results, uploads, and production receipts." + version = "0.1.0" config_model = QuoteJudgeStudioV1Config - auth_model = PlatformUserAuth + auth_model = {{ auth_type }} + # Hosted generated agents read the caller's saved LLM credential through + # ctx.llm. The platform may proxy that credential through LiteLLM, but agent + # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY + # directly. + llm_provisioning = LLMProvisioning.PLATFORM pricing = Pricing( price_per_call_usd=0.0, - caller_pays_llm=False, - notes="Deterministic quote scoring. No LLM credential is required.", + caller_pays_llm=True, + notes="Starter agent uses the caller's saved LLM credential via ctx.llm.", ) - resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120) workspace_access = WorkspaceAccess.dynamic( - max_files=8, + max_files=64, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=False, - max_total_size_bytes=512 * 1024, ) - tools_used = ("postgres", "mcp", "packed-frontend") - platform_resources = AgentPlatformResources( - databases=( - AgentDatabase( - name="quote-judge-studio-v1-data", - scope="user", - access_mode="read_write", - env=AgentDatabaseEnv(url=DATABASE_ENV), - migrations=AgentDatabaseMigrations(path="db/migrations"), - ), + tools_used = ("deepagents", "langchain") + + @a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful") + async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str: + creds = ctx.llm + await ctx.emit_progress(f"llm: {creds.model} via {creds.source}") + if not creds.api_key: + return ( + "LLM key required. Add an LLM credential in Settings > LLM " + "credentials before running this agent; for local --invoke " + "runs set AGENT_LLM_KEY." + ) + graph = self._build_deep_agent(ctx=ctx, creds=creds) + state = await graph.ainvoke( + {"messages": [{"role": "user", "content": prompt}]}, + config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT}, ) - ) + await ctx.emit_progress("deepagent finished") + return _last_message_text(state) - @a2a.tool( - description=( - "Compare at least two vendor quotes with price/delivery/warranty " - "weights, persist the result, and return a recommendation." - ), - timeout_seconds=60, - idempotent=True, - cost_class="deterministic", - ) - async def compare_quotes( + def _build_deep_agent( self, - ctx: RunContext[PlatformUserAuth], - comparison_id: Annotated[str, Field(min_length=1, max_length=120)], - quotes: Annotated[list[QuoteInput], Field(max_length=MAX_QUOTES)], - weights: QuoteWeights, - ) -> dict[str, Any]: - await ctx.emit_progress("Validating quote comparison inputs") - tenant = _tenant_key(ctx) - result = _build_comparison_result(comparison_id, quotes, weights) - if not result["ok"]: - await _persist_receipt(ctx, tenant, comparison_id, "compare_quotes", "validation_error", {"code": result["code"]}, result) - return result + *, + ctx: RunContext[{{ auth_type }}], + creds: LLMCreds, + ) -> Any: + # Lazy imports keep `a2a card` usable before local dependencies are + # installed. `a2a deploy` installs requirements.txt during the build. + from a2a_pack.deepagents import create_a2a_deep_agent + from langchain.agents.middleware import wrap_model_call + from langchain_core.tools import tool - await ctx.emit_progress("Persisting comparison and production receipt") - await _save_comparison(tenant, comparison_id, result) - receipt = await _persist_receipt(ctx, tenant, comparison_id, "compare_quotes", "ok", _safe_input_payload(comparison_id, quotes, weights), result) - result["receipt"] = receipt - return result + @tool + def text_stats(text: str) -> str: + """Return exact word, character, and line counts for text.""" + words = [part for part in text.split() if part.strip()] + return json.dumps( + { + "characters": len(text), + "words": len(words), + "lines": len(text.splitlines()) or 1, + } + ) - @a2a.tool( - description="Reopen a previously persisted quote comparison by id.", - timeout_seconds=30, - idempotent=True, - cost_class="deterministic", - ) - async def get_comparison( - self, - ctx: RunContext[PlatformUserAuth], - comparison_id: Annotated[str, Field(min_length=1, max_length=120)], - ) -> dict[str, Any]: - tenant = _tenant_key(ctx) - record = await _load_comparison(tenant, comparison_id) - if record is None: - result = { - "ok": False, - "code": "comparison_not_found", - "message": "No comparison was found for this id in your workspace.", - "comparison_id": comparison_id, - } - await _persist_receipt(ctx, tenant, comparison_id, "get_comparison", "not_found", {"comparison_id": comparison_id}, result) - return result - await _persist_receipt(ctx, tenant, comparison_id, "get_comparison", "ok", {"comparison_id": comparison_id}, record) - return record + @wrap_model_call + async def log_model_call(request: Any, handler: Any) -> Any: + messages = request.state.get("messages", []) + print( + "[middleware] model_call " + f"model={creds.model} source={creds.source} messages={len(messages)}" + ) + return await handler(request) - @a2a.tool( - description="List recently saved quote comparisons for the signed-in user.", - timeout_seconds=30, - idempotent=True, - cost_class="deterministic", - ) - async def list_comparisons( - self, - ctx: RunContext[PlatformUserAuth], - limit: Annotated[int, Field(ge=1, le=50)] = 10, - ) -> dict[str, Any]: - rows = await _list_comparisons(_tenant_key(ctx), limit) - return {"ok": True, "comparisons": rows} - - @a2a.tool( - description=( - "Browser-safe upload bridge: accept bounded base64 JSON/CSV/text " - "quote files, parse them, compare, persist, and return the result." - ), - timeout_seconds=60, - idempotent=True, - cost_class="deterministic", - ) - async def compare_quotes_from_browser_upload( - self, - ctx: RunContext[PlatformUserAuth], - comparison_id: Annotated[str, Field(min_length=1, max_length=120)], - documents: Annotated[list[BrowserDocument], Field(min_length=1, max_length=MAX_BROWSER_DOCUMENTS)], - weights: QuoteWeights, - ) -> dict[str, Any]: - parsed = _parse_browser_documents(documents) - if not parsed["ok"]: - await _persist_receipt(ctx, _tenant_key(ctx), comparison_id, "compare_quotes_from_browser_upload", "validation_error", {"comparison_id": comparison_id}, parsed) - return {**parsed, "comparison_id": comparison_id} - return await self.compare_quotes(ctx, comparison_id=comparison_id, quotes=parsed["quotes"], weights=weights) - - @a2a.tool( - description="Compatibility alias for the original QuoteJudge browser upload tool.", - timeout_seconds=60, - idempotent=True, - cost_class="deterministic", - ) - async def compare_uploaded_quotes_browser( - self, - ctx: RunContext[PlatformUserAuth], - comparison_id: Annotated[str, Field(min_length=1, max_length=120)], - documents: Annotated[list[BrowserDocument], Field(min_length=1, max_length=MAX_BROWSER_DOCUMENTS)], - weights: QuoteWeights, - ) -> dict[str, Any]: - return await self.compare_quotes_from_browser_upload( + backend = ctx.workspace_backend() + skill_sources = _seed_runtime_skills(backend, ctx) + # create_a2a_deep_agent resolves provider:model strings with + # langchain.init_chat_model from ctx.llm, preserving LiteLLM routing, + # provider-specific extra body, and runtime model overrides. + return create_a2a_deep_agent( ctx, - comparison_id=comparison_id, - documents=documents, - weights=weights, - ) - - @a2a.tool( - description=( - "External API upload path: accept a typed FileUpload containing " - "JSON, CSV, or pasted text quote data and run the same comparison." - ), - timeout_seconds=60, - idempotent=True, - cost_class="deterministic", - grant_mode="read_only", - grant_allow_patterns=("**",), - ) - async def compare_quotes_from_upload( - self, - ctx: RunContext[PlatformUserAuth], - comparison_id: Annotated[str, Field(min_length=1, max_length=120)], - quote_file: Annotated[ - UploadedFile, - FileUpload( - accept=ACCEPTED_UPLOAD_TYPES, - max_bytes=MAX_BROWSER_DOCUMENT_BYTES, - description="JSON list/object, CSV, or text containing vendor quotes.", - ), - ], - weights: QuoteWeights, - ) -> dict[str, Any]: - if quote_file.size_bytes > MAX_BROWSER_DOCUMENT_BYTES: - return _validation_error( - "file_too_large", - f"Upload is {quote_file.size_bytes} bytes; maximum is {MAX_BROWSER_DOCUMENT_BYTES} bytes.", - comparison_id=comparison_id, - ) - if quote_file.media_type not in ACCEPTED_UPLOAD_TYPES: - return _validation_error( - "unsupported_media_type", - f"Unsupported media type {quote_file.media_type!r}. Use JSON, CSV, or plain text.", - comparison_id=comparison_id, - ) - reader = getattr(ctx.workspace, "read_bytes", None) - if reader is None: - return _validation_error("workspace_read_unavailable", "The runtime did not expose uploaded file bytes.", comparison_id=comparison_id) - try: - raw = reader(quote_file.path) - except Exception: # noqa: BLE001 - return _validation_error("file_read_failed", "Could not read the uploaded file from the granted workspace.", comparison_id=comparison_id) - parsed = _parse_document_bytes(quote_file.filename, quote_file.media_type, raw) - if not parsed["ok"]: - return {**parsed, "comparison_id": comparison_id} - return await self.compare_quotes(ctx, comparison_id=comparison_id, quotes=parsed["quotes"], weights=weights) - - @a2a.tool( - description="Compatibility alias for the original QuoteJudge external upload tool.", - timeout_seconds=60, - idempotent=True, - cost_class="deterministic", - grant_mode="read_only", - grant_allow_patterns=("**",), - ) - async def compare_uploaded_quote_file( - self, - ctx: RunContext[PlatformUserAuth], - comparison_id: Annotated[str, Field(min_length=1, max_length=120)], - document: Annotated[ - UploadedFile, - FileUpload( - accept=ACCEPTED_UPLOAD_TYPES, - max_bytes=MAX_BROWSER_DOCUMENT_BYTES, - description="JSON list/object, CSV, or text containing vendor quotes.", - ), - ], - weights: QuoteWeights, - ) -> dict[str, Any]: - return await self.compare_quotes_from_upload( - ctx, - comparison_id=comparison_id, - quote_file=document, - weights=weights, + creds=creds, + backend=backend, + skills=skill_sources or None, + tools=[text_stats], + middleware=[log_model_call], + system_prompt=SYSTEM_PROMPT, ) -def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str: - auth = ctx.auth - stable = auth.user_id if auth.user_id is not None else (auth.sub or auth.email) - return f"user:{stable}" +def _runtime_skills_root(ctx: RunContext[Any]) -> str: + workspace = getattr(ctx, "_workspace", None) + prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ()) + if not prefixes: + outputs_prefix = getattr(workspace, "outputs_prefix", None) + prefixes = (outputs_prefix or "outputs/",) + prefix = str(prefixes[0]).strip("/") + return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}" -def _safe_input_payload(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]: - return { - "comparison_id": comparison_id, - "quote_count": len(quotes), - "vendors": [quote.vendor for quote in quotes], - "weights": weights.model_dump(mode="json"), - } +def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]: + """Copy packaged DeepAgents skills into the invocation workspace. + + DeepAgents loads skills from its backend, while source-controlled + ``skills/`` folders live in the image. This bridge lets generated agents + ship reusable SKILL.md bundles without giving up durable A2A workspace + files. + """ + root = Path(__file__).parent / "skills" + if not root.exists(): + return [] + runtime_skills_root = _runtime_skills_root(ctx) + uploads: list[tuple[str, bytes]] = [] + for path in root.rglob("*"): + if path.is_file(): + rel = path.relative_to(root).as_posix() + uploads.append((runtime_skills_root + rel, path.read_bytes())) + if uploads: + backend.upload_files(uploads) + return [runtime_skills_root] + return [] -def _validation_error(code: str, message: str, *, comparison_id: str | None = None, details: list[dict[str, Any]] | None = None) -> dict[str, Any]: - payload: dict[str, Any] = { - "ok": False, - "code": code, - "message": message, - "validation_errors": details or [{"field": "quotes", "message": message}], - } - if comparison_id is not None: - payload["comparison_id"] = comparison_id - return payload +def _last_message_text(state: dict[str, Any]) -> str: + messages = state.get("messages") or [] + if not messages: + return json.dumps(state, default=str) - -def _build_comparison_result(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]: - if len(quotes) < 2: - return _validation_error( - "at_least_two_quotes_required", - "Provide at least two vendor quotes before requesting a recommendation.", - comparison_id=comparison_id, - details=[{"field": "quotes", "message": "At least two quotes are required."}], - ) - if len(quotes) > MAX_QUOTES: - return _validation_error( - "too_many_quotes", - f"Provide no more than {MAX_QUOTES} quotes.", - comparison_id=comparison_id, - ) - total_weight = weights.price + weights.delivery + weights.warranty - if total_weight <= 0: - return _validation_error( - "weights_must_sum_positive", - "At least one of price, delivery, or warranty weight must be greater than zero.", - comparison_id=comparison_id, - details=[{"field": "weights", "message": "Weights must sum to a positive number."}], - ) - - min_price = min(q.unit_price for q in quotes) - min_delivery = min(q.delivery_days for q in quotes) - max_warranty = max(q.warranty_months for q in quotes) - normalized_weights = { - "price": weights.price / total_weight, - "delivery": weights.delivery / total_weight, - "warranty": weights.warranty / total_weight, - } - - rows: list[dict[str, Any]] = [] - for quote in quotes: - price_score = (min_price / quote.unit_price) * 100 if quote.unit_price else 0.0 - delivery_score = 100.0 if quote.delivery_days == 0 and min_delivery == 0 else ((min_delivery + 1) / (quote.delivery_days + 1)) * 100 - warranty_score = (quote.warranty_months / max_warranty) * 100 if max_warranty else 100.0 - weighted_score = ( - normalized_weights["price"] * price_score - + normalized_weights["delivery"] * delivery_score - + normalized_weights["warranty"] * warranty_score - ) - total_price = quote.unit_price * quote.quantity - rows.append( - { - "vendor": quote.vendor, - "unit_price": round(quote.unit_price, 4), - "quantity": quote.quantity, - "total_price": round(total_price, 2), - "delivery_days": quote.delivery_days, - "warranty_months": quote.warranty_months, - "scores": { - "price": round(price_score, 2), - "delivery": round(delivery_score, 2), - "warranty": round(warranty_score, 2), - "weighted_total": round(weighted_score, 2), - }, - } - ) - - rows.sort(key=_quote_sort_key) - winner = rows[0] - recommendation = { - "vendor": winner["vendor"], - "score": winner["scores"]["weighted_total"], - "reason": ( - f"{winner['vendor']} has the strongest weighted score " - f"({winner['scores']['weighted_total']:.2f}) using price={weights.price:g}, " - f"delivery={weights.delivery:g}, warranty={weights.warranty:g}." - ), - } - return { - "ok": True, - "comparison_id": comparison_id, - "recommendation": recommendation, - "weights": { - "price": weights.price, - "delivery": weights.delivery, - "warranty": weights.warranty, - "normalized": {key: round(value, 4) for key, value in normalized_weights.items()}, - }, - "quote_count": len(quotes), - "comparison_table": rows, - "created_at": datetime.now(timezone.utc).isoformat(), - } - - -def _quote_sort_key(item: dict[str, Any]) -> tuple[Any, ...]: - return ( - -item["scores"]["weighted_total"], - item["total_price"], - item["delivery_days"], - item["vendor"].lower(), - ) - - -def _parse_browser_documents(documents: list[BrowserDocument]) -> dict[str, Any]: - quotes: list[QuoteInput] = [] - errors: list[dict[str, Any]] = [] - for index, doc in enumerate(documents): - if doc.media_type not in ACCEPTED_UPLOAD_TYPES: - errors.append({"field": f"documents[{index}].media_type", "message": "Use JSON, CSV, or plain text."}) - continue - try: - raw = base64.b64decode(doc.data_base64, validate=True) - except Exception: # noqa: BLE001 - errors.append({"field": f"documents[{index}].data_base64", "message": "Invalid base64 payload."}) - continue - if len(raw) > MAX_BROWSER_DOCUMENT_BYTES: - errors.append({"field": f"documents[{index}]", "message": f"File exceeds {MAX_BROWSER_DOCUMENT_BYTES} bytes."}) - continue - parsed = _parse_document_bytes(doc.filename, doc.media_type, raw) - if parsed["ok"]: - quotes.extend(parsed["quotes"]) - else: - errors.extend(parsed.get("validation_errors") or []) - if errors: - return _validation_error("upload_parse_failed", "One or more uploaded quote files could not be parsed.", details=errors) - return {"ok": True, "quotes": quotes[:MAX_QUOTES]} - - -def _parse_document_bytes(filename: str, media_type: str, raw: bytes) -> dict[str, Any]: - if len(raw) > MAX_BROWSER_DOCUMENT_BYTES: - return _validation_error("file_too_large", f"{filename} exceeds the {MAX_BROWSER_DOCUMENT_BYTES} byte limit.") - text = raw.decode("utf-8", errors="replace").strip() - if not text: - return _validation_error("empty_upload", "The uploaded quote file was empty.") - try: - if media_type == "application/json" or filename.lower().endswith(".json"): - data = json.loads(text) - items = data.get("quotes") if isinstance(data, dict) else data - if not isinstance(items, list): - return _validation_error("invalid_json_quotes", "JSON must be a list of quotes or an object with a quotes list.") - return {"ok": True, "quotes": [QuoteInput.model_validate(item) for item in items]} - if media_type == "text/csv" or filename.lower().endswith(".csv"): - reader = csv.DictReader(io.StringIO(text)) - return {"ok": True, "quotes": [QuoteInput.model_validate(row) for row in reader]} - return {"ok": True, "quotes": _parse_loose_text_quotes(text)} - except Exception as exc: # noqa: BLE001 - return _validation_error("quote_parse_failed", f"Could not parse quote data: {type(exc).__name__}.") - - -def _parse_loose_text_quotes(text: str) -> list[QuoteInput]: - quotes: list[QuoteInput] = [] - for line in text.splitlines(): - parts = [part.strip() for part in re.split(r"[,|;]\s*", line) if part.strip()] - if len(parts) < 5 or parts[0].lower() in {"vendor", "supplier"}: - continue - quotes.append( - QuoteInput( - vendor=parts[0], - unit_price=float(parts[1]), - quantity=int(float(parts[2])), - delivery_days=int(float(parts[3])), - warranty_months=int(float(parts[4])), - ) - ) - if not quotes: - raise ValueError("No quote rows found. Expected vendor, unit_price, quantity, delivery_days, warranty_months.") - return quotes - - -def _db_url() -> str: - return os.environ.get(DATABASE_ENV, "").strip() - - -def _connect(): - url = _db_url() - if not url: - raise RuntimeError("managed database is not configured") - import psycopg - - return psycopg.connect(url, autocommit=True) - - -async def _save_comparison(tenant_key: str, comparison_id: str, result: dict[str, Any]) -> None: - with _connect() as conn: - with conn.cursor() as cur: - cur.execute( - """ - INSERT INTO quote_comparisons ( - tenant_key, comparison_id, recommendation_vendor, recommended_vendor, - result, payload, weighted_score, score, normalized_total - ) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) - ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET - recommendation_vendor = EXCLUDED.recommendation_vendor, - recommended_vendor = EXCLUDED.recommended_vendor, - result = EXCLUDED.result, - payload = EXCLUDED.payload, - weighted_score = EXCLUDED.weighted_score, - score = EXCLUDED.score, - normalized_total = EXCLUDED.normalized_total, - updated_at = NOW() - """, - ( - tenant_key, - comparison_id, - result["recommendation"]["vendor"], - result["recommendation"]["vendor"], - json.dumps(result), - json.dumps(result), - result["recommendation"]["score"], - result["recommendation"]["score"], - result["recommendation"]["score"], - ), - ) - - -async def _load_comparison(tenant_key: str, comparison_id: str) -> dict[str, Any] | None: - with _connect() as conn: - with conn.cursor() as cur: - cur.execute( - "SELECT COALESCE(result, payload) FROM quote_comparisons WHERE tenant_key = %s AND comparison_id = %s", - (tenant_key, comparison_id), - ) - row = cur.fetchone() - if row is None: - return None - value = row[0] - if isinstance(value, str): - return json.loads(value) - return value - - -async def _list_comparisons(tenant_key: str, limit: int) -> list[dict[str, Any]]: - with _connect() as conn: - with conn.cursor() as cur: - cur.execute( - """ - SELECT comparison_id, COALESCE(recommendation_vendor, recommended_vendor), updated_at, COALESCE(result, payload) - FROM quote_comparisons - WHERE tenant_key = %s - ORDER BY updated_at DESC - LIMIT %s - """, - (tenant_key, limit), - ) - rows = cur.fetchall() - output: list[dict[str, Any]] = [] - for comparison_id, recommendation_vendor, updated_at, result in rows: - output.append( - { - "comparison_id": comparison_id, - "recommendation_vendor": recommendation_vendor, - "updated_at": updated_at.isoformat() if hasattr(updated_at, "isoformat") else str(updated_at), - "quote_count": (result or {}).get("quote_count") if isinstance(result, dict) else None, - } - ) - return output - - -async def _persist_receipt( - ctx: RunContext[PlatformUserAuth], - tenant_key: str, - comparison_id: str, - tool_name: str, - status: str, - inputs: dict[str, Any], - result: dict[str, Any], -) -> dict[str, Any]: - payload = { - "agent": QuoteJudgeStudioV1.name, - "version": QuoteJudgeStudioV1.version, - "tool": tool_name, - "status": status, - "comparison_id": comparison_id, - "caller_scope": "platform_user", - "task_id": getattr(ctx, "task_id", ""), - "input_hash": _hash_json(inputs), - "result_hash": _hash_json(_receipt_safe_result(result)), - "created_at": datetime.now(timezone.utc).isoformat(), - } - payload["receipt_id"] = "qjr_" + _hash_json(payload)[:24] - try: - with _connect() as conn: - with conn.cursor() as cur: - cur.execute( - """ - INSERT INTO quote_judge_receipts - (receipt_id, tenant_key, comparison_id, tool_name, status, input_hash, payload) - VALUES (%s, %s, %s, %s, %s, %s, %s) - ON CONFLICT (receipt_id) DO NOTHING - """, - (payload["receipt_id"], tenant_key, comparison_id, tool_name, status, payload["input_hash"], json.dumps(payload)), - ) - except Exception as exc: # noqa: BLE001 - await ctx.emit_error("Receipt persistence failed; comparison result was still computed.", code="receipt_persist_failed") - return {"persisted": False, "code": "receipt_persist_failed", "message": type(exc).__name__} - return {"persisted": True, "receipt_id": payload["receipt_id"], "input_hash": payload["input_hash"]} - - -def _receipt_safe_result(result: dict[str, Any]) -> dict[str, Any]: - return { - "ok": result.get("ok"), - "comparison_id": result.get("comparison_id"), - "recommendation": result.get("recommendation"), - "code": result.get("code"), - } - - -def _hash_json(value: Any) -> str: - raw = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") - return hashlib.sha256(raw).hexdigest() + content = getattr(messages[-1], "content", None) + if isinstance(content, str): + return content + if isinstance(content, list): + parts: list[str] = [] + for item in content: + if isinstance(item, dict): + text = item.get("text") or item.get("content") + if text: + parts.append(str(text)) + elif item: + parts.append(str(item)) + return "\n".join(parts) if parts else json.dumps(content, default=str) + return str(content or messages[-1])