a2a-source-edit: write agent.py
This commit is contained in:
a2a-cloud
730
agent.py
730
agent.py
@@ -1,639 +1,189 @@
|
|||||||
"""QuoteJudge Studio full-stack A2A agent.
|
"""quote-judge-studio-v1 agent.
|
||||||
|
|
||||||
Deterministic, no-LLM quote comparison product:
|
Starter stack:
|
||||||
- platform-authenticated user tenancy;
|
- DeepAgents for tool-calling orchestration
|
||||||
- managed Postgres persistence;
|
- Caller-provided LLM credentials via ctx.llm
|
||||||
- typed A2A/MCP tools for comparison, reload, browser uploads, API uploads;
|
- A tiny model-call middleware hook you can replace with tracing,
|
||||||
- production receipt records without secrets.
|
routing, rate limits, or policy checks
|
||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import base64
|
|
||||||
import csv
|
|
||||||
import hashlib
|
|
||||||
import io
|
|
||||||
import json
|
import json
|
||||||
import os
|
from pathlib import Path
|
||||||
import re
|
from typing import Any
|
||||||
from datetime import datetime, timezone
|
|
||||||
from typing import Annotated, Any
|
|
||||||
|
|
||||||
from pydantic import BaseModel, Field, field_validator
|
from pydantic import BaseModel
|
||||||
|
|
||||||
import a2a_pack as a2a
|
import a2a_pack as a2a
|
||||||
from a2a_pack import (
|
from a2a_pack import (
|
||||||
A2AAgent,
|
A2AAgent,
|
||||||
AgentDatabase,
|
LLMProvisioning,
|
||||||
AgentDatabaseEnv,
|
{{ auth_type }},
|
||||||
AgentDatabaseMigrations,
|
|
||||||
AgentPlatformResources,
|
|
||||||
FileUpload,
|
|
||||||
PlatformUserAuth,
|
|
||||||
Pricing,
|
Pricing,
|
||||||
Resources,
|
|
||||||
RunContext,
|
RunContext,
|
||||||
UploadedFile,
|
|
||||||
WorkspaceAccess,
|
WorkspaceAccess,
|
||||||
WorkspaceMode,
|
WorkspaceMode,
|
||||||
)
|
)
|
||||||
|
from a2a_pack.context import LLMCreds
|
||||||
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"
|
|
||||||
|
|
||||||
|
|
||||||
class QuoteJudgeStudioV1Config(BaseModel):
|
class QuoteJudgeStudioV1Config(BaseModel):
|
||||||
"""No user-editable config is required."""
|
pass
|
||||||
|
|
||||||
|
|
||||||
class QuoteInput(BaseModel):
|
SYSTEM_PROMPT = """\
|
||||||
vendor: Annotated[str, Field(min_length=1, max_length=160)]
|
You are a compact tool-calling agent.
|
||||||
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)]
|
|
||||||
|
|
||||||
@field_validator("vendor")
|
Use the text_stats tool when the user asks about text, counts, summaries,
|
||||||
@classmethod
|
or anything where exact length/word numbers would help. Mention tool results
|
||||||
def _clean_vendor(cls, value: str) -> str:
|
briefly instead of dumping raw JSON.
|
||||||
cleaned = re.sub(r"\s+", " ", value).strip()
|
"""
|
||||||
if not cleaned:
|
|
||||||
raise ValueError("vendor is required")
|
RUNTIME_SKILLS_DIR = "quote-judge-studio-v1/.deepagents/skills/"
|
||||||
return cleaned
|
DEEPAGENTS_RECURSION_LIMIT = 500
|
||||||
|
|
||||||
|
|
||||||
class QuoteWeights(BaseModel):
|
class QuoteJudgeStudioV1(A2AAgent[QuoteJudgeStudioV1Config, {{ auth_type }}]):
|
||||||
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]):
|
|
||||||
name = "quote-judge-studio-v1"
|
name = "quote-judge-studio-v1"
|
||||||
description = (
|
description = "One-page QuoteJudge studio for comparing vendor quotes with weighted price, delivery, and warranty scoring, persistent results, uploads, and production receipts."
|
||||||
"QuoteJudge compares vendor quotes with weighted price, delivery, "
|
version = "0.1.0"
|
||||||
"and warranty scoring, persists recommendations, supports upload/paste "
|
|
||||||
"workflows, and exposes typed A2A/MCP tools."
|
|
||||||
)
|
|
||||||
version = "0.1.2"
|
|
||||||
|
|
||||||
config_model = QuoteJudgeStudioV1Config
|
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(
|
pricing = Pricing(
|
||||||
price_per_call_usd=0.0,
|
price_per_call_usd=0.0,
|
||||||
caller_pays_llm=False,
|
caller_pays_llm=True,
|
||||||
notes="Deterministic quote scoring. No LLM credential is required.",
|
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(
|
workspace_access = WorkspaceAccess.dynamic(
|
||||||
max_files=8,
|
max_files=64,
|
||||||
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
|
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
|
||||||
require_reason=False,
|
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(
|
@a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful")
|
||||||
description=(
|
async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str:
|
||||||
"Compare at least two vendor quotes with price/delivery/warranty "
|
creds = ctx.llm
|
||||||
"weights, persist the result, and return a recommendation."
|
await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
|
||||||
),
|
if not creds.api_key:
|
||||||
timeout_seconds=60,
|
|
||||||
idempotent=True,
|
|
||||||
cost_class="deterministic",
|
|
||||||
)
|
|
||||||
async def compare_quotes(
|
|
||||||
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
|
|
||||||
|
|
||||||
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
|
|
||||||
|
|
||||||
@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
|
|
||||||
|
|
||||||
@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(
|
|
||||||
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,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
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 _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 _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 _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 (
|
return (
|
||||||
-item["scores"]["weighted_total"],
|
"LLM key required. Add an LLM credential in Settings > LLM "
|
||||||
item["total_price"],
|
"credentials before running this agent; for local --invoke "
|
||||||
item["delivery_days"],
|
"runs set AGENT_LLM_KEY."
|
||||||
item["vendor"].lower(),
|
|
||||||
)
|
)
|
||||||
|
graph = self._build_deep_agent(ctx=ctx, creds=creds)
|
||||||
|
state = await graph.ainvoke(
|
||||||
def _parse_browser_documents(documents: list[BrowserDocument]) -> dict[str, Any]:
|
{"messages": [{"role": "user", "content": prompt}]},
|
||||||
quotes: list[QuoteInput] = []
|
config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT},
|
||||||
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])),
|
|
||||||
)
|
)
|
||||||
)
|
await ctx.emit_progress("deepagent finished")
|
||||||
if not quotes:
|
return _last_message_text(state)
|
||||||
raise ValueError("No quote rows found. Expected vendor, unit_price, quantity, delivery_days, warranty_months.")
|
|
||||||
return quotes
|
|
||||||
|
|
||||||
|
def _build_deep_agent(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
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
|
||||||
|
|
||||||
def _db_url() -> str:
|
@tool
|
||||||
return os.environ.get(DATABASE_ENV, "").strip()
|
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()]
|
||||||
def _connect():
|
return json.dumps(
|
||||||
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,
|
"characters": len(text),
|
||||||
"recommendation_vendor": recommendation_vendor,
|
"words": len(words),
|
||||||
"updated_at": updated_at.isoformat() if hasattr(updated_at, "isoformat") else str(updated_at),
|
"lines": len(text.splitlines()) or 1,
|
||||||
"quote_count": (result or {}).get("quote_count") if isinstance(result, dict) else None,
|
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
return output
|
|
||||||
|
@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,
|
||||||
|
creds=creds,
|
||||||
|
backend=backend,
|
||||||
|
skills=skill_sources or None,
|
||||||
|
tools=[text_stats],
|
||||||
|
middleware=[log_model_call],
|
||||||
|
system_prompt=SYSTEM_PROMPT,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
async def _persist_receipt(
|
def _runtime_skills_root(ctx: RunContext[Any]) -> str:
|
||||||
ctx: RunContext[PlatformUserAuth],
|
workspace = getattr(ctx, "_workspace", None)
|
||||||
tenant_key: str,
|
prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
|
||||||
comparison_id: str,
|
if not prefixes:
|
||||||
tool_name: str,
|
outputs_prefix = getattr(workspace, "outputs_prefix", None)
|
||||||
status: str,
|
prefixes = (outputs_prefix or "outputs/",)
|
||||||
inputs: dict[str, Any],
|
prefix = str(prefixes[0]).strip("/")
|
||||||
result: dict[str, Any],
|
return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
|
||||||
) -> dict[str, Any]:
|
|
||||||
payload = {
|
|
||||||
"agent": QuoteJudgeStudioV1.name,
|
def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
|
||||||
"version": QuoteJudgeStudioV1.version,
|
"""Copy packaged DeepAgents skills into the invocation workspace.
|
||||||
"tool": tool_name,
|
|
||||||
"status": status,
|
DeepAgents loads skills from its backend, while source-controlled
|
||||||
"comparison_id": comparison_id,
|
``skills/`` folders live in the image. This bridge lets generated agents
|
||||||
"caller_scope": "platform_user",
|
ship reusable SKILL.md bundles without giving up durable A2A workspace
|
||||||
"task_id": getattr(ctx, "task_id", ""),
|
files.
|
||||||
"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
|
root = Path(__file__).parent / "skills"
|
||||||
(receipt_id, tenant_key, comparison_id, tool_name, status, input_hash, payload)
|
if not root.exists():
|
||||||
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
return []
|
||||||
ON CONFLICT (receipt_id) DO NOTHING
|
runtime_skills_root = _runtime_skills_root(ctx)
|
||||||
""",
|
uploads: list[tuple[str, bytes]] = []
|
||||||
(payload["receipt_id"], tenant_key, comparison_id, tool_name, status, payload["input_hash"], json.dumps(payload)),
|
for path in root.rglob("*"):
|
||||||
)
|
if path.is_file():
|
||||||
except Exception as exc: # noqa: BLE001
|
rel = path.relative_to(root).as_posix()
|
||||||
await ctx.emit_error("Receipt persistence failed; comparison result was still computed.", code="receipt_persist_failed")
|
uploads.append((runtime_skills_root + rel, path.read_bytes()))
|
||||||
return {"persisted": False, "code": "receipt_persist_failed", "message": type(exc).__name__}
|
if uploads:
|
||||||
return {"persisted": True, "receipt_id": payload["receipt_id"], "input_hash": payload["input_hash"]}
|
backend.upload_files(uploads)
|
||||||
|
return [runtime_skills_root]
|
||||||
|
return []
|
||||||
|
|
||||||
|
|
||||||
def _receipt_safe_result(result: dict[str, Any]) -> dict[str, Any]:
|
def _last_message_text(state: dict[str, Any]) -> str:
|
||||||
return {
|
messages = state.get("messages") or []
|
||||||
"ok": result.get("ok"),
|
if not messages:
|
||||||
"comparison_id": result.get("comparison_id"),
|
return json.dumps(state, default=str)
|
||||||
"recommendation": result.get("recommendation"),
|
|
||||||
"code": result.get("code"),
|
|
||||||
}
|
|
||||||
|
|
||||||
|
content = getattr(messages[-1], "content", None)
|
||||||
def _hash_json(value: Any) -> str:
|
if isinstance(content, str):
|
||||||
raw = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
|
return content
|
||||||
return hashlib.sha256(raw).hexdigest()
|
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