a2a-source-edit: write agent.py
This commit is contained in:
a2a-cloud
864
agent.py
864
agent.py
@@ -1,769 +1,189 @@
|
||||
"""QuoteJudge Studio v1: deterministic quote comparison product agent.
|
||||
"""quote-judge-studio-v1 agent.
|
||||
|
||||
This agent intentionally does not call an LLM. It exposes a compact typed A2A
|
||||
surface, a packed React app, managed Postgres persistence, MCP-compatible tools,
|
||||
and bounded upload paths for both browser JSON and external FileUpload clients.
|
||||
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 binascii
|
||||
import csv
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from datetime import UTC, datetime
|
||||
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,
|
||||
RunContext,
|
||||
UploadedFile,
|
||||
WorkspaceAccess,
|
||||
WorkspaceMode,
|
||||
)
|
||||
|
||||
DATABASE_NAME = "quote-judge-studio-v1-data"
|
||||
MAX_QUOTES = 20
|
||||
MAX_UPLOAD_FILES = 2
|
||||
MAX_UPLOAD_BYTES = 64 * 1024
|
||||
MAX_BASE64_CHARS = ((MAX_UPLOAD_BYTES + 2) // 3) * 4 + 4
|
||||
ALLOWED_UPLOAD_MEDIA_TYPES = {
|
||||
"application/json",
|
||||
"text/json",
|
||||
"text/csv",
|
||||
"application/csv",
|
||||
"text/plain",
|
||||
"application/octet-stream",
|
||||
}
|
||||
_COMPARISON_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,79}$")
|
||||
_LOCAL_COMPARISONS: dict[tuple[str, str], dict[str, Any]] = {}
|
||||
_LOCAL_RECEIPTS: list[dict[str, Any]] = []
|
||||
from a2a_pack.context import LLMCreds
|
||||
|
||||
|
||||
class QuoteJudgeStudioV1Config(BaseModel):
|
||||
pass
|
||||
|
||||
|
||||
class QuoteInput(BaseModel):
|
||||
vendor: str = Field(min_length=1, max_length=120)
|
||||
unit_price: float = Field(gt=0, le=1_000_000)
|
||||
quantity: int = Field(gt=0, le=10_000_000)
|
||||
delivery_days: int = Field(ge=0, le=3650)
|
||||
warranty_months: 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 = " ".join(str(value).strip().split())
|
||||
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: float = Field(ge=0, le=1000)
|
||||
delivery: float = Field(ge=0, le=1000)
|
||||
warranty: float = Field(ge=0, le=1000)
|
||||
|
||||
|
||||
class BrowserDocument(BaseModel):
|
||||
filename: str = Field(min_length=1, max_length=160)
|
||||
media_type: str = Field(min_length=1, max_length=120)
|
||||
data_base64: str = Field(min_length=1, max_length=MAX_BASE64_CHARS)
|
||||
|
||||
|
||||
QuoteList = Annotated[list[QuoteInput], Field(max_length=MAX_QUOTES)]
|
||||
DocumentList = Annotated[list[BrowserDocument], Field(min_length=1, max_length=MAX_UPLOAD_FILES)]
|
||||
ComparisonId = Annotated[str, Field(min_length=1, max_length=80)]
|
||||
|
||||
|
||||
class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, PlatformUserAuth]):
|
||||
class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, {{ auth_type }}]):
|
||||
name = "quote-judge-studio-v1"
|
||||
description = (
|
||||
"QuoteJudge Studio compares vendor quotes with weighted price, delivery, "
|
||||
"and warranty scoring, persists results, supports bounded uploads, and "
|
||||
"exposes typed MCP-compatible A2A tools."
|
||||
)
|
||||
version = "0.1.1"
|
||||
description = "QuoteJudge compares vendor quotes with weighted price, delivery, and warranty scoring, persists per-user comparisons, supports browser uploads, and serves a polished packed product UI."
|
||||
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 and managed Postgres persistence; no LLM required.",
|
||||
caller_pays_llm=True,
|
||||
notes="Starter agent uses the caller's saved LLM credential via ctx.llm.",
|
||||
)
|
||||
workspace_access = WorkspaceAccess.dynamic(
|
||||
max_files=4,
|
||||
allowed_modes=(WorkspaceMode.READ_ONLY,),
|
||||
max_files=64,
|
||||
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
|
||||
require_reason=False,
|
||||
max_total_size_bytes=MAX_UPLOAD_BYTES * MAX_UPLOAD_FILES,
|
||||
)
|
||||
platform_resources = AgentPlatformResources(
|
||||
databases=(
|
||||
AgentDatabase(
|
||||
name=DATABASE_NAME,
|
||||
scope="user",
|
||||
access_mode="read_write",
|
||||
env=AgentDatabaseEnv(url="DATABASE_URL"),
|
||||
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},
|
||||
)
|
||||
)
|
||||
tools_used = ("postgres", "mcp", "file-upload")
|
||||
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 and persist a production receipt.",
|
||||
timeout_seconds=30,
|
||||
cost_class="deterministic",
|
||||
)
|
||||
async def compare_quotes(
|
||||
def _build_deep_agent(
|
||||
self,
|
||||
ctx: RunContext[PlatformUserAuth],
|
||||
comparison_id: ComparisonId,
|
||||
quotes: QuoteList,
|
||||
weights: QuoteWeights,
|
||||
) -> dict[str, Any]:
|
||||
tenant = _tenant_key(ctx)
|
||||
validation = _validate_compare_inputs(comparison_id, quotes, weights)
|
||||
if validation is not None:
|
||||
await _persist_receipt(ctx, tenant, "compare_quotes", {"comparison_id": comparison_id}, validation)
|
||||
return validation
|
||||
*,
|
||||
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
|
||||
|
||||
result = _score_quotes(comparison_id, quotes, weights)
|
||||
result["tenant_scope"] = "user"
|
||||
result["receipt"] = await _persist_receipt(
|
||||
@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,
|
||||
}
|
||||
)
|
||||
|
||||
@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)
|
||||
|
||||
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,
|
||||
tenant,
|
||||
"compare_quotes",
|
||||
{
|
||||
"comparison_id": comparison_id,
|
||||
"quote_count": len(quotes),
|
||||
"weights": weights.model_dump(mode="json"),
|
||||
},
|
||||
result,
|
||||
creds=creds,
|
||||
backend=backend,
|
||||
skills=skill_sources or None,
|
||||
tools=[text_stats],
|
||||
middleware=[log_model_call],
|
||||
system_prompt=SYSTEM_PROMPT,
|
||||
)
|
||||
saved = await _save_comparison(tenant, result)
|
||||
if not saved["ok"]:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": saved["code"],
|
||||
"message": saved["message"],
|
||||
"comparison_id": comparison_id,
|
||||
"action": "Retry after the managed database is available, or contact the agent owner.",
|
||||
}
|
||||
await ctx.emit_progress(f"saved comparison {comparison_id}")
|
||||
return result
|
||||
|
||||
@a2a.tool(
|
||||
description="Reopen a persisted QuoteJudge comparison by id.",
|
||||
timeout_seconds=15,
|
||||
cost_class="deterministic",
|
||||
idempotent=True,
|
||||
)
|
||||
async def get_comparison(
|
||||
self,
|
||||
ctx: RunContext[PlatformUserAuth],
|
||||
comparison_id: ComparisonId,
|
||||
) -> dict[str, Any]:
|
||||
tenant = _tenant_key(ctx)
|
||||
if not _valid_comparison_id(comparison_id):
|
||||
return _validation_error(
|
||||
comparison_id,
|
||||
"invalid_comparison_id",
|
||||
"comparison_id must start with a letter or number and contain only letters, numbers, dash, underscore, dot, or colon.",
|
||||
)
|
||||
result = await _load_comparison(tenant, comparison_id)
|
||||
if result is None:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "comparison_not_found",
|
||||
"comparison_id": comparison_id,
|
||||
"message": "No saved comparison exists for this signed-in user and comparison_id.",
|
||||
"action": "Run compare_quotes first, then call get_comparison with the same comparison_id.",
|
||||
}
|
||||
return result
|
||||
|
||||
@a2a.tool(
|
||||
description="Compatibility browser upload bridge: parse bounded base64 JSON/CSV quote files, compare, persist, and receipt the result.",
|
||||
timeout_seconds=30,
|
||||
cost_class="deterministic",
|
||||
)
|
||||
async def upload_quotes(
|
||||
self,
|
||||
ctx: RunContext[PlatformUserAuth],
|
||||
comparison_id: ComparisonId,
|
||||
documents: DocumentList,
|
||||
weights: QuoteWeights,
|
||||
) -> dict[str, Any]:
|
||||
parsed = _parse_browser_documents(documents)
|
||||
if not parsed["ok"]:
|
||||
await _persist_receipt(ctx, _tenant_key(ctx), "upload_quotes", {"comparison_id": comparison_id}, parsed)
|
||||
return parsed | {"comparison_id": comparison_id}
|
||||
return await self.compare_quotes(ctx, comparison_id, parsed["quotes"], weights)
|
||||
|
||||
@a2a.tool(
|
||||
description="Compatibility external FileUpload tool: parse one bounded JSON/CSV quote file, compare, persist, and receipt the result.",
|
||||
timeout_seconds=30,
|
||||
cost_class="deterministic",
|
||||
)
|
||||
async def upload_quote_file(
|
||||
self,
|
||||
ctx: RunContext[PlatformUserAuth],
|
||||
comparison_id: ComparisonId,
|
||||
file: Annotated[
|
||||
UploadedFile,
|
||||
FileUpload(
|
||||
accept=tuple(sorted(ALLOWED_UPLOAD_MEDIA_TYPES)),
|
||||
max_bytes=MAX_UPLOAD_BYTES,
|
||||
description="JSON or CSV quote file with vendor, unit_price, quantity, delivery_days, warranty_months.",
|
||||
),
|
||||
],
|
||||
weights: QuoteWeights,
|
||||
) -> dict[str, Any]:
|
||||
if file.size_bytes > MAX_UPLOAD_BYTES:
|
||||
return _validation_error(
|
||||
comparison_id,
|
||||
"upload_too_large",
|
||||
f"Upload is {file.size_bytes} bytes; maximum is {MAX_UPLOAD_BYTES} bytes.",
|
||||
)
|
||||
if file.media_type not in ALLOWED_UPLOAD_MEDIA_TYPES:
|
||||
return _validation_error(
|
||||
comparison_id,
|
||||
"unsupported_media_type",
|
||||
"Upload must be JSON, CSV, or plain text containing JSON/CSV quote data.",
|
||||
)
|
||||
try:
|
||||
reader = getattr(ctx.workspace, "read_bytes", None)
|
||||
if reader is None:
|
||||
raise RuntimeError("workspace reader unavailable")
|
||||
data = reader(file.path)
|
||||
except Exception: # noqa: BLE001
|
||||
return _validation_error(
|
||||
comparison_id,
|
||||
"upload_unreadable",
|
||||
"The uploaded file could not be read from the granted workspace. Re-upload the file and retry.",
|
||||
)
|
||||
if len(data) > MAX_UPLOAD_BYTES:
|
||||
return _validation_error(
|
||||
comparison_id,
|
||||
"upload_too_large",
|
||||
f"Decoded upload is {len(data)} bytes; maximum is {MAX_UPLOAD_BYTES} bytes.",
|
||||
)
|
||||
parsed = _parse_quote_bytes(data, filename=file.filename, media_type=file.media_type)
|
||||
if not parsed["ok"]:
|
||||
return parsed | {"comparison_id": comparison_id}
|
||||
return await self.compare_quotes(ctx, comparison_id, parsed["quotes"], weights)
|
||||
|
||||
|
||||
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 _valid_comparison_id(value: str) -> bool:
|
||||
return bool(_COMPARISON_ID_RE.fullmatch(str(value or "").strip()))
|
||||
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 _validate_compare_inputs(
|
||||
comparison_id: str,
|
||||
quotes: list[QuoteInput],
|
||||
weights: QuoteWeights,
|
||||
) -> dict[str, Any] | None:
|
||||
if not _valid_comparison_id(comparison_id):
|
||||
return _validation_error(
|
||||
comparison_id,
|
||||
"invalid_comparison_id",
|
||||
"comparison_id must start with a letter or number and contain only letters, numbers, dash, underscore, dot, or colon.",
|
||||
)
|
||||
if len(quotes) < 2:
|
||||
return _validation_error(
|
||||
comparison_id,
|
||||
"at_least_two_quotes_required",
|
||||
"QuoteJudge needs at least two vendor quotes to make a comparison.",
|
||||
)
|
||||
if len({quote.vendor.casefold() for quote in quotes}) < 2:
|
||||
return _validation_error(
|
||||
comparison_id,
|
||||
"at_least_two_vendors_required",
|
||||
"Provide quotes from at least two distinct vendors.",
|
||||
)
|
||||
total_weight = weights.price + weights.delivery + weights.warranty
|
||||
if total_weight <= 0:
|
||||
return _validation_error(
|
||||
comparison_id,
|
||||
"positive_weight_required",
|
||||
"At least one of price, delivery, or warranty weight must be greater than zero.",
|
||||
)
|
||||
return None
|
||||
def _last_message_text(state: dict[str, Any]) -> str:
|
||||
messages = state.get("messages") or []
|
||||
if not messages:
|
||||
return json.dumps(state, default=str)
|
||||
|
||||
|
||||
def _validation_error(comparison_id: str, code: str, message: str) -> dict[str, Any]:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": code,
|
||||
"comparison_id": comparison_id,
|
||||
"message": message,
|
||||
"action": "Correct the input and retry the same tool call.",
|
||||
}
|
||||
|
||||
|
||||
def _score_quotes(comparison_id: str, quotes: list[QuoteInput], weights: QuoteWeights) -> dict[str, Any]:
|
||||
rows = [
|
||||
{
|
||||
**quote.model_dump(mode="json"),
|
||||
"total_price": round(quote.unit_price * quote.quantity, 4),
|
||||
}
|
||||
for quote in quotes
|
||||
]
|
||||
price_values = [row["total_price"] for row in rows]
|
||||
delivery_values = [row["delivery_days"] for row in rows]
|
||||
warranty_values = [row["warranty_months"] for row in rows]
|
||||
total_weight = weights.price + weights.delivery + weights.warranty
|
||||
|
||||
scored: list[dict[str, Any]] = []
|
||||
for row in rows:
|
||||
price_score = _lower_is_better(row["total_price"], min(price_values), max(price_values))
|
||||
delivery_score = _lower_is_better(row["delivery_days"], min(delivery_values), max(delivery_values))
|
||||
warranty_score = _higher_is_better(row["warranty_months"], min(warranty_values), max(warranty_values))
|
||||
weighted_score = (
|
||||
price_score * weights.price
|
||||
+ delivery_score * weights.delivery
|
||||
+ warranty_score * weights.warranty
|
||||
) / total_weight
|
||||
scored.append(
|
||||
{
|
||||
**row,
|
||||
"scores": {
|
||||
"price": round(price_score, 2),
|
||||
"delivery": round(delivery_score, 2),
|
||||
"warranty": round(warranty_score, 2),
|
||||
"weighted_total": round(weighted_score, 2),
|
||||
},
|
||||
}
|
||||
)
|
||||
scored.sort(
|
||||
key=lambda row: (
|
||||
-row["scores"]["weighted_total"],
|
||||
row["total_price"],
|
||||
row["delivery_days"],
|
||||
-row["warranty_months"],
|
||||
row["vendor"].casefold(),
|
||||
)
|
||||
)
|
||||
winner = scored[0]
|
||||
now = datetime.now(UTC).isoformat()
|
||||
return {
|
||||
"ok": True,
|
||||
"comparison_id": comparison_id,
|
||||
"recommendation": {
|
||||
"vendor": winner["vendor"],
|
||||
"score": winner["scores"]["weighted_total"],
|
||||
"reason": (
|
||||
f"{winner['vendor']} has the strongest weighted score using "
|
||||
f"price={weights.price:g}, delivery={weights.delivery:g}, warranty={weights.warranty:g}."
|
||||
),
|
||||
},
|
||||
"weights": weights.model_dump(mode="json"),
|
||||
"quote_count": len(quotes),
|
||||
"quotes": scored,
|
||||
"created_at": now,
|
||||
"updated_at": now,
|
||||
}
|
||||
|
||||
|
||||
def _lower_is_better(value: float, min_value: float, max_value: float) -> float:
|
||||
if max_value == min_value:
|
||||
return 100.0
|
||||
return max(0.0, min(100.0, 100.0 * (max_value - value) / (max_value - min_value)))
|
||||
|
||||
|
||||
def _higher_is_better(value: float, min_value: float, max_value: float) -> float:
|
||||
if max_value == min_value:
|
||||
return 100.0
|
||||
return max(0.0, min(100.0, 100.0 * (value - min_value) / (max_value - min_value)))
|
||||
|
||||
|
||||
def _parse_browser_documents(documents: list[BrowserDocument]) -> dict[str, Any]:
|
||||
all_quotes: list[QuoteInput] = []
|
||||
for document in documents:
|
||||
filename = _safe_filename(document.filename)
|
||||
media_type = document.media_type.strip().lower()
|
||||
if media_type not in ALLOWED_UPLOAD_MEDIA_TYPES:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "unsupported_media_type",
|
||||
"message": "Upload must be JSON, CSV, or plain text containing JSON/CSV quote data.",
|
||||
"action": "Upload a .json or .csv file exported from your quote sheet.",
|
||||
}
|
||||
raw_b64 = document.data_base64.strip()
|
||||
if len(raw_b64) > MAX_BASE64_CHARS:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "upload_too_large",
|
||||
"message": f"{filename} exceeds the {MAX_UPLOAD_BYTES} byte decoded upload limit.",
|
||||
"action": "Trim the file or paste the quotes directly into the form.",
|
||||
}
|
||||
try:
|
||||
data = base64.b64decode(raw_b64, validate=True)
|
||||
except (binascii.Error, ValueError):
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "invalid_base64",
|
||||
"message": f"{filename} is not valid base64 data.",
|
||||
"action": "Re-select the file in the browser and retry the upload.",
|
||||
}
|
||||
if len(data) > MAX_UPLOAD_BYTES:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "upload_too_large",
|
||||
"message": f"{filename} decoded to {len(data)} bytes; maximum is {MAX_UPLOAD_BYTES} bytes.",
|
||||
"action": "Trim the file or paste the quotes directly into the form.",
|
||||
}
|
||||
parsed = _parse_quote_bytes(data, filename=filename, media_type=media_type)
|
||||
if not parsed["ok"]:
|
||||
return parsed
|
||||
all_quotes.extend(parsed["quotes"])
|
||||
if len(all_quotes) > MAX_QUOTES:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "too_many_quotes",
|
||||
"message": f"At most {MAX_QUOTES} quotes can be compared in one call.",
|
||||
"action": "Split the quote set into smaller comparisons.",
|
||||
}
|
||||
return {"ok": True, "quotes": all_quotes}
|
||||
|
||||
|
||||
def _safe_filename(value: str) -> str:
|
||||
cleaned = re.sub(r"[^A-Za-z0-9._ -]", "_", str(value or "upload").strip())[:160]
|
||||
return cleaned or "upload"
|
||||
|
||||
|
||||
def _parse_quote_bytes(data: bytes, *, filename: str, media_type: str) -> dict[str, Any]:
|
||||
try:
|
||||
text = data.decode("utf-8-sig")
|
||||
except UnicodeDecodeError:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "upload_not_utf8",
|
||||
"message": f"{filename} must be UTF-8 encoded JSON or CSV text.",
|
||||
"action": "Export the file as UTF-8 CSV or JSON and retry.",
|
||||
}
|
||||
if media_type in {"application/json", "text/json"} or filename.lower().endswith(".json"):
|
||||
return _parse_json_quotes(text, filename=filename)
|
||||
return _parse_csv_quotes(text, filename=filename)
|
||||
|
||||
|
||||
def _parse_json_quotes(text: str, *, filename: str) -> dict[str, Any]:
|
||||
try:
|
||||
payload = json.loads(text)
|
||||
except json.JSONDecodeError as exc:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "invalid_json",
|
||||
"message": f"{filename} is not valid JSON: {exc.msg}.",
|
||||
"action": "Use a JSON array of quote objects, or an object with a quotes array.",
|
||||
}
|
||||
raw_quotes = payload.get("quotes") if isinstance(payload, dict) else payload
|
||||
if not isinstance(raw_quotes, list):
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "quotes_array_required",
|
||||
"message": f"{filename} must contain a quotes array.",
|
||||
"action": "Provide [{vendor, unit_price, quantity, delivery_days, warranty_months}, ...].",
|
||||
}
|
||||
return _coerce_quotes(raw_quotes, source=filename)
|
||||
|
||||
|
||||
def _parse_csv_quotes(text: str, *, filename: str) -> dict[str, Any]:
|
||||
try:
|
||||
rows = list(csv.DictReader(io.StringIO(text)))
|
||||
except csv.Error as exc:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "invalid_csv",
|
||||
"message": f"{filename} is not valid CSV: {exc}.",
|
||||
"action": "Use a header row with vendor, unit_price, quantity, delivery_days, warranty_months.",
|
||||
}
|
||||
if not rows:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "no_quotes_found",
|
||||
"message": f"{filename} did not contain any quote rows.",
|
||||
"action": "Add at least two vendor quote rows.",
|
||||
}
|
||||
return _coerce_quotes(rows, source=filename)
|
||||
|
||||
|
||||
def _coerce_quotes(raw_quotes: list[Any], *, source: str) -> dict[str, Any]:
|
||||
if len(raw_quotes) > MAX_QUOTES:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "too_many_quotes",
|
||||
"message": f"{source} contains {len(raw_quotes)} quotes; maximum is {MAX_QUOTES}.",
|
||||
"action": "Split the file into smaller comparisons.",
|
||||
}
|
||||
quotes: list[QuoteInput] = []
|
||||
errors: list[str] = []
|
||||
for idx, item in enumerate(raw_quotes, start=1):
|
||||
if not isinstance(item, dict):
|
||||
errors.append(f"row {idx}: expected an object")
|
||||
continue
|
||||
try:
|
||||
quotes.append(QuoteInput.model_validate(item))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
errors.append(f"row {idx}: {exc}")
|
||||
if errors:
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "invalid_quote_rows",
|
||||
"message": "One or more quote rows are invalid.",
|
||||
"errors": errors[:5],
|
||||
"action": "Fix the listed rows and retry.",
|
||||
}
|
||||
return {"ok": True, "quotes": quotes}
|
||||
|
||||
|
||||
async def _save_comparison(tenant: str, result: dict[str, Any]) -> dict[str, Any]:
|
||||
db_url = os.environ.get("DATABASE_URL")
|
||||
key = (tenant, result["comparison_id"])
|
||||
if not db_url:
|
||||
_LOCAL_COMPARISONS[key] = json.loads(json.dumps(result, default=str))
|
||||
return {"ok": True}
|
||||
try:
|
||||
import psycopg
|
||||
from psycopg.types.json import Jsonb
|
||||
|
||||
with psycopg.connect(db_url, autocommit=True) as conn:
|
||||
_ensure_schema(conn)
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO quote_judge_comparisons
|
||||
(tenant_key, comparison_id, recommendation_vendor, result_json, updated_at)
|
||||
VALUES (%s, %s, %s, %s, NOW())
|
||||
ON CONFLICT (tenant_key, comparison_id) DO UPDATE SET
|
||||
recommendation_vendor = EXCLUDED.recommendation_vendor,
|
||||
result_json = EXCLUDED.result_json,
|
||||
updated_at = NOW()
|
||||
""",
|
||||
(
|
||||
tenant,
|
||||
result["comparison_id"],
|
||||
result["recommendation"]["vendor"],
|
||||
Jsonb(result),
|
||||
),
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO app_records (tenant_key, record_kind, payload, updated_at)
|
||||
VALUES (%s, 'quote_comparison', %s, NOW())
|
||||
""",
|
||||
(tenant, Jsonb(result)),
|
||||
)
|
||||
return {"ok": True}
|
||||
except Exception: # noqa: BLE001
|
||||
return {
|
||||
"ok": False,
|
||||
"code": "database_unavailable",
|
||||
"message": "The managed Postgres database could not save this comparison.",
|
||||
}
|
||||
|
||||
|
||||
async def _load_comparison(tenant: str, comparison_id: str) -> dict[str, Any] | None:
|
||||
key = (tenant, comparison_id)
|
||||
db_url = os.environ.get("DATABASE_URL")
|
||||
if not db_url:
|
||||
return _LOCAL_COMPARISONS.get(key)
|
||||
try:
|
||||
import psycopg
|
||||
|
||||
with psycopg.connect(db_url) as conn:
|
||||
_ensure_schema(conn)
|
||||
row = conn.execute(
|
||||
"""
|
||||
SELECT result_json FROM quote_judge_comparisons
|
||||
WHERE tenant_key = %s AND comparison_id = %s
|
||||
ORDER BY updated_at DESC LIMIT 1
|
||||
""",
|
||||
(tenant, comparison_id),
|
||||
).fetchone()
|
||||
if row and row[0]:
|
||||
return dict(row[0])
|
||||
row = conn.execute(
|
||||
"""
|
||||
SELECT payload FROM app_records
|
||||
WHERE tenant_key = %s
|
||||
AND record_kind IN ('quote_comparison', 'comparison', 'quotejudge_comparison')
|
||||
AND payload->>'comparison_id' = %s
|
||||
ORDER BY updated_at DESC LIMIT 1
|
||||
""",
|
||||
(tenant, comparison_id),
|
||||
).fetchone()
|
||||
if row and row[0]:
|
||||
return dict(row[0])
|
||||
return _load_legacy_comparison(conn, tenant, comparison_id)
|
||||
except Exception: # noqa: BLE001
|
||||
return None
|
||||
|
||||
|
||||
def _ensure_schema(conn: Any) -> None:
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS app_records (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
tenant_key TEXT NOT NULL,
|
||||
record_kind TEXT NOT NULL,
|
||||
payload JSONB NOT NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
)
|
||||
"""
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS quote_judge_comparisons (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
tenant_key TEXT NOT NULL,
|
||||
comparison_id TEXT NOT NULL,
|
||||
recommendation_vendor TEXT NOT NULL,
|
||||
result_json JSONB NOT NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
UNIQUE (tenant_key, comparison_id)
|
||||
)
|
||||
"""
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS quote_judge_receipts (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
receipt_id TEXT,
|
||||
tenant_key TEXT NOT NULL,
|
||||
comparison_id TEXT,
|
||||
skill_name TEXT,
|
||||
tool_name TEXT,
|
||||
status TEXT,
|
||||
input_hash TEXT,
|
||||
result_hash TEXT,
|
||||
receipt_json JSONB,
|
||||
payload JSONB,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
)
|
||||
"""
|
||||
)
|
||||
conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS receipt_id TEXT")
|
||||
conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS skill_name TEXT")
|
||||
conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS tool_name TEXT")
|
||||
conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS status TEXT")
|
||||
conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS input_hash TEXT")
|
||||
conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS result_hash TEXT")
|
||||
conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS receipt_json JSONB")
|
||||
conn.execute("ALTER TABLE quote_judge_receipts ADD COLUMN IF NOT EXISTS payload JSONB")
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS app_records_tenant_kind_idx ON app_records (tenant_key, record_kind, created_at DESC)")
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS quote_judge_receipts_tenant_idx ON quote_judge_receipts (tenant_key, created_at DESC)")
|
||||
|
||||
|
||||
def _load_legacy_comparison(conn: Any, tenant: str, comparison_id: str) -> dict[str, Any] | None:
|
||||
for table in ("quote_comparisons", "comparisons"):
|
||||
columns = _table_columns(conn, table)
|
||||
if not columns or "tenant_key" not in columns or "comparison_id" not in columns:
|
||||
continue
|
||||
for json_col in ("result_json", "payload", "result", "data"):
|
||||
if json_col not in columns:
|
||||
continue
|
||||
row = conn.execute(
|
||||
f"SELECT {json_col} FROM {table} WHERE tenant_key = %s AND comparison_id = %s LIMIT 1",
|
||||
(tenant, comparison_id),
|
||||
).fetchone()
|
||||
if row and isinstance(row[0], dict):
|
||||
return dict(row[0])
|
||||
return None
|
||||
|
||||
|
||||
def _table_columns(conn: Any, table: str) -> set[str]:
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT column_name FROM information_schema.columns
|
||||
WHERE table_schema = current_schema() AND table_name = %s
|
||||
""",
|
||||
(table,),
|
||||
).fetchall()
|
||||
return {str(row[0]) for row in rows}
|
||||
|
||||
|
||||
async def _persist_receipt(
|
||||
ctx: RunContext[PlatformUserAuth],
|
||||
tenant: str,
|
||||
skill_name: str,
|
||||
inputs: dict[str, Any],
|
||||
result: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
comparison_id = str(inputs.get("comparison_id") or result.get("comparison_id") or "")
|
||||
receipt = {
|
||||
"receipt_type": "quote_judge_execution",
|
||||
"agent": QuoteJudgeStudioV1.name,
|
||||
"agent_version": QuoteJudgeStudioV1.version,
|
||||
"skill_name": skill_name,
|
||||
"tool_name": skill_name,
|
||||
"comparison_id": comparison_id,
|
||||
"tenant_scope": "user",
|
||||
"task_id": getattr(ctx, "task_id", ""),
|
||||
"created_at": datetime.now(UTC).isoformat(),
|
||||
"input_hash": _hash_json(inputs),
|
||||
"result_hash": _hash_json({k: v for k, v in result.items() if k != "receipt"}),
|
||||
"status": "ok" if result.get("ok") else "validation_error",
|
||||
}
|
||||
receipt["receipt_id"] = _hash_json(receipt)[:32]
|
||||
db_url = os.environ.get("DATABASE_URL")
|
||||
if not db_url:
|
||||
_LOCAL_RECEIPTS.append(receipt)
|
||||
return {"persisted": True, "receipt_id": receipt["receipt_id"]}
|
||||
try:
|
||||
import psycopg
|
||||
from psycopg.types.json import Jsonb
|
||||
|
||||
with psycopg.connect(db_url, autocommit=True) as conn:
|
||||
_ensure_schema(conn)
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO quote_judge_receipts
|
||||
(receipt_id, tenant_key, comparison_id, skill_name, tool_name,
|
||||
status, input_hash, result_hash, receipt_json, payload)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||||
""",
|
||||
(
|
||||
receipt["receipt_id"],
|
||||
tenant,
|
||||
comparison_id or None,
|
||||
skill_name,
|
||||
skill_name,
|
||||
receipt["status"],
|
||||
receipt["input_hash"],
|
||||
receipt["result_hash"],
|
||||
Jsonb(receipt),
|
||||
Jsonb(receipt),
|
||||
),
|
||||
)
|
||||
return {"persisted": True, "receipt_id": receipt["receipt_id"]}
|
||||
except Exception: # noqa: BLE001
|
||||
return {"persisted": False, "receipt_id": receipt["receipt_id"], "warning": "receipt_database_write_failed"}
|
||||
|
||||
|
||||
def _hash_json(payload: dict[str, Any]) -> str:
|
||||
data = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
|
||||
return hashlib.sha256(data).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])
|
||||
|
||||
Reference in New Issue
Block a user