bobseares-2/contract-clock-studio-v1
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

a2a-source-edit: write agent.py

Browse Source
This commit is contained in:
a2a-cloud
2026-07-18 08:08:41 +00:00
parent 11e097a1a5
commit 9283156ce7
1 changed files with 155 additions and 476 deletions
Download Patch File Download Diff File Expand all files Collapse all files

631
agent.py
View File

@@ -1,510 +1,189 @@
"""ContractClock Studio full-stack A2A agent. """contract-clock-studio-v1 agent.
Deterministic product backend for the ContractClock one-page startup. The agent Starter stack:
extracts only explicit dated renewal obligations and notice windows, persists - DeepAgents for tool-calling orchestration
contract timelines in user-scoped managed Postgres, and exposes the same typed - Caller-provided LLM credentials via ctx.llm
A2A tools to Agent API clients, MCP clients, and the packed React frontend. - A tiny model-call middleware hook you can replace with tracing,
routing, rate limits, or policy checks
""" """
from __future__ import annotations from __future__ import annotations
import asyncio import json
import base64 from pathlib import Path
import binascii from typing import Any
import hashlib
import os
import re
from datetime import date, datetime, timedelta, timezone
from typing import Annotated, Any
from pydantic import BaseModel, Field 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,
State, WorkspaceAccess,
UploadedFile, WorkspaceMode,
)
MAX_CONTRACT_TEXT_CHARS = 60_000
MAX_UPLOAD_BYTES = 512_000
ALLOWED_BROWSER_MEDIA_TYPES = (
"text/plain",
"text/markdown",
"application/octet-stream",
)
DATE_RE = re.compile(r"\b(20\d{2}-\d{2}-\d{2})\b")
RENEWAL_DATE_RE = re.compile(
r"\brenew(?:s|al|ed|ing)?\b[^.\n;]{0,160}?\b(?:on|as\s+of|effective(?:\s+on)?|"
r"date(?:\s+is)?|until|through)\b\s*:?\s*(20\d{2}-\d{2}-\d{2})\b",
re.IGNORECASE,
)
RENEWAL_RE = re.compile(r"\brenew(?:s|al|ed|ing)?\b", re.IGNORECASE)
NOTICE_RE = re.compile(
r"(?:at\s+least\s+)?(?P<days>\d{1,3})\s+days?\s+(?:written\s+)?notice",
re.IGNORECASE,
)
AMBIGUOUS_DATE_RE = re.compile(
r"\b(sometime|spring|summer|fall|autumn|winter|soon|later|well\s+in\s+advance|next\s+\w+)\b",
re.IGNORECASE,
) )
from a2a_pack.context import LLMCreds
class ContractClockStudioV1Config(BaseModel): class ContractClockStudioV1Config(BaseModel):
pass pass
class BrowserDocument(BaseModel): SYSTEM_PROMPT = """\
filename: Annotated[str, Field(min_length=1, max_length=160)] You are a compact tool-calling agent.
media_type: Annotated[str, Field(min_length=1, max_length=120)]
data_base64: Annotated[str, Field(min_length=1, max_length=700_000)] 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 = "contract-clock-studio-v1/.deepagents/skills/"
DEEPAGENTS_RECURSION_LIMIT = 500
class ContractClockStudioV1(A2AAgent[ContractClockStudioV1Config, PlatformUserAuth]): class ContractClockStudioV1(A2AAgent[ContractClockStudioV1Config, {{ auth_type }}]):
name = "contract-clock-studio-v1" name = "contract-clock-studio-v1"
description = ( description = "ContractClock: upload or paste contracts, extract explicit renewal and notice deadlines, persist user-scoped timelines, and reopen them in a packed React app."
"ContractClock extracts explicit renewal and notice deadlines from " version = "0.1.0"
"contracts, saves user-scoped timelines, and reopens them in a packed "
"one-page studio."
)
version = "0.1.5"
config_model = ContractClockStudioV1Config config_model = ContractClockStudioV1Config
auth_model = PlatformUserAuth auth_model = {{ auth_type }}
state = State.DURABLE # Hosted generated agents read the caller's saved LLM credential through
state_model = ContractClockStudioV1Config # ctx.llm. The platform may proxy that credential through LiteLLM, but agent
resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=120) # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY
platform_resources = AgentPlatformResources( # directly.
databases=( llm_provisioning = LLMProvisioning.PLATFORM
AgentDatabase(
name="contract-clock-studio-v1-data",
scope="user",
access_mode="read_write",
env=AgentDatabaseEnv(url="DATABASE_URL"),
migrations=AgentDatabaseMigrations(path="db/migrations"),
),
)
)
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 local extraction; no LLM credential required.", notes="Starter agent uses the caller's saved LLM credential via ctx.llm.",
) )
tools_used = ("postgres", "mcp", "packed-frontend") workspace_access = WorkspaceAccess.dynamic(
max_files=64,
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
require_reason=False,
)
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:
"Analyze pasted contract text, extract explicit renewal and notice " creds = ctx.llm
"deadlines, atomically persist the contract timeline and execution receipt." await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
), if not creds.api_key:
timeout_seconds=30, return (
idempotent=False, "LLM key required. Add an LLM credential in Settings > LLM "
cost_class="cheap", "credentials before running this agent; for local --invoke "
) "runs set AGENT_LLM_KEY."
async def analyze_contract(
self,
ctx: RunContext[PlatformUserAuth],
contract_id: Annotated[str, Field(min_length=1, max_length=120)],
text: Annotated[str, Field(min_length=1, max_length=MAX_CONTRACT_TEXT_CHARS)],
title: Annotated[str, Field(min_length=1, max_length=200)] = "Untitled contract",
) -> dict[str, Any]:
tenant = _tenant_key(ctx)
if tenant is None:
return _auth_failure()
analysis = _analyze_text(contract_id=contract_id, title=title, text=text)
if not analysis["ok"]:
return analysis
try:
await _save_analysis(
tenant_key=tenant,
contract_id=contract_id,
title=title,
text=text,
analysis=analysis,
skill_name="analyze_contract",
task_id=getattr(ctx, "task_id", ""),
) )
except DatabaseUnavailable as exc: graph = self._build_deep_agent(ctx=ctx, creds=creds)
return _db_failure(exc) state = await graph.ainvoke(
return analysis {"messages": [{"role": "user", "content": prompt}]},
config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT},
@a2a.tool( )
description="Reload a previously analyzed contract deadline timeline from user-scoped storage.", await ctx.emit_progress("deepagent finished")
timeout_seconds=20, return _last_message_text(state)
idempotent=True,
cost_class="cheap", def _build_deep_agent(
) self,
async def get_contract_deadlines( *,
self, ctx: RunContext[{{ auth_type }}],
ctx: RunContext[PlatformUserAuth], creds: LLMCreds,
contract_id: Annotated[str, Field(min_length=1, max_length=120)], ) -> Any:
) -> dict[str, Any]: # Lazy imports keep `a2a card` usable before local dependencies are
tenant = _tenant_key(ctx) # installed. `a2a deploy` installs requirements.txt during the build.
if tenant is None: from a2a_pack.deepagents import create_a2a_deep_agent
return _auth_failure() from langchain.agents.middleware import wrap_model_call
try: from langchain_core.tools import tool
record = await _load_contract(tenant_key=tenant, contract_id=contract_id)
except DatabaseUnavailable as exc: @tool
return _db_failure(exc) def text_stats(text: str) -> str:
if record is None: """Return exact word, character, and line counts for text."""
return {"ok": False, "code": "not_found", "contract_id": contract_id} words = [part for part in text.split() if part.strip()]
return { return json.dumps(
"ok": True, {
"contract_id": contract_id, "characters": len(text),
"title": record.get("title") or "Untitled contract", "words": len(words),
"deadlines": record.get("deadlines") or [], "lines": len(text.splitlines()) or 1,
"receipt": record.get("receipt") or {}, }
"updated_at": record.get("updated_at"), )
}
@wrap_model_call
@a2a.tool( async def log_model_call(request: Any, handler: Any) -> Any:
description=( messages = request.state.get("messages", [])
"Analyze a text contract uploaded by an external Agent API client. " print(
"The FileUpload schema is retained for typed multipart clients." "[middleware] model_call "
), f"model={creds.model} source={creds.source} messages={len(messages)}"
timeout_seconds=30, )
idempotent=False, return await handler(request)
cost_class="cheap",
) backend = ctx.workspace_backend()
async def analyze_contract_file( skill_sources = _seed_runtime_skills(backend, ctx)
self, # create_a2a_deep_agent resolves provider:model strings with
ctx: RunContext[PlatformUserAuth], # langchain.init_chat_model from ctx.llm, preserving LiteLLM routing,
contract_id: Annotated[str, Field(min_length=1, max_length=120)], # provider-specific extra body, and runtime model overrides.
document: Annotated[ return create_a2a_deep_agent(
UploadedFile, ctx,
FileUpload( creds=creds,
accept=ALLOWED_BROWSER_MEDIA_TYPES, backend=backend,
max_bytes=MAX_UPLOAD_BYTES, skills=skill_sources or None,
description="UTF-8 text contract document", tools=[text_stats],
), middleware=[log_model_call],
], system_prompt=SYSTEM_PROMPT,
title: Annotated[str, Field(min_length=1, max_length=200)] = "Uploaded contract",
) -> dict[str, Any]:
try:
payload = _read_uploaded_file(ctx, document)
except UploadRejected as exc:
return exc.result
return await self.analyze_contract(ctx, contract_id=contract_id, text=payload, title=title)
@a2a.tool(
description=(
"Browser-safe upload bridge accepting one bounded base64 JSON text "
"document from the packed frontend."
),
timeout_seconds=30,
idempotent=False,
cost_class="cheap",
)
async def analyze_contract_upload(
self,
ctx: RunContext[PlatformUserAuth],
contract_id: Annotated[str, Field(min_length=1, max_length=120)],
document: BrowserDocument,
title: Annotated[str, Field(min_length=1, max_length=200)] = "Uploaded contract",
) -> dict[str, Any]:
try:
payload = _decode_browser_document(document)
except UploadRejected as exc:
return exc.result
return await self.analyze_contract(ctx, contract_id=contract_id, text=payload, title=title)
class DatabaseUnavailable(RuntimeError):
def __init__(self, code: str, message: str) -> None:
self.code = code
super().__init__(message)
class UploadRejected(ValueError):
def __init__(self, code: str, message: str) -> None:
self.result = {"ok": False, "code": code, "message": message}
super().__init__(message)
def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str | None:
auth = getattr(ctx, "auth", None)
user_id = getattr(auth, "user_id", None)
sub = (getattr(auth, "sub", None) or "").strip()
if user_id is not None:
return f"user:{user_id}"
if sub:
return f"user:{sub}"
return None
def _auth_failure() -> dict[str, Any]:
return {
"ok": False,
"code": "auth_required",
"message": "A signed-in A2A Cloud user session is required to use ContractClock.",
}
def _db_failure(exc: DatabaseUnavailable) -> dict[str, Any]:
return {
"ok": False,
"code": exc.code,
"message": "ContractClock storage is not ready. Retry after the managed database is available.",
}
def _analyze_text(*, contract_id: str, title: str, text: str) -> dict[str, Any]:
clean_text = (text or "").strip()
if not clean_text:
return {"ok": False, "code": "empty_contract", "contract_id": contract_id}
if len(clean_text) > MAX_CONTRACT_TEXT_CHARS:
return {"ok": False, "code": "text_too_large", "contract_id": contract_id}
explicit_dates = []
for match in DATE_RE.finditer(clean_text):
try:
explicit_dates.append(date.fromisoformat(match.group(1)))
except ValueError:
return {"ok": False, "code": "invalid_date", "contract_id": contract_id}
if not explicit_dates:
code = "ambiguous_date" if AMBIGUOUS_DATE_RE.search(clean_text) else "missing_explicit_date"
return {"ok": False, "code": code, "contract_id": contract_id}
deadlines: list[dict[str, str]] = []
renewal_match = RENEWAL_DATE_RE.search(clean_text)
if renewal_match:
renewal_date = date.fromisoformat(renewal_match.group(1))
deadlines.append({"kind": "renewal", "date": renewal_date.isoformat()})
notice_match = NOTICE_RE.search(clean_text)
if notice_match:
notice_days = int(notice_match.group("days"))
if 1 <= notice_days <= 366:
deadlines.append(
{"kind": "notice", "date": (renewal_date - timedelta(days=notice_days)).isoformat()}
)
elif RENEWAL_RE.search(clean_text):
return {
"ok": False,
"code": "ambiguous_renewal_date",
"contract_id": contract_id,
}
else:
# Explicit dates outside a renewal clause are kept as dated obligations;
# relative invoice terms without a concrete anchor are intentionally not guessed.
for item in explicit_dates:
deadlines.append({"kind": "obligation", "date": item.isoformat()})
if not deadlines:
return {"ok": False, "code": "no_supported_deadline", "contract_id": contract_id}
return {
"ok": True,
"contract_id": contract_id,
"title": title,
"deadlines": deadlines,
"rationale": (
"Only ISO-formatted explicit dates were extracted. Notice deadlines "
"are calculated from explicit renewal dates and written-notice day counts."
),
}
def _decode_browser_document(document: BrowserDocument) -> str:
filename = document.filename.strip()
if "/" in filename or "\\" in filename or filename in {".", ".."}:
raise UploadRejected("invalid_filename", "Upload filename must be a simple file name.")
media_type = document.media_type.split(";", 1)[0].strip().lower()
if media_type not in ALLOWED_BROWSER_MEDIA_TYPES:
raise UploadRejected("unsupported_media_type", "Upload a UTF-8 text contract file.")
try:
raw = base64.b64decode(document.data_base64, validate=True)
except (binascii.Error, ValueError) as exc:
raise UploadRejected("invalid_base64", "Upload data must be valid base64.") from exc
return _decode_uploaded_bytes(raw)
def _read_uploaded_file(ctx: RunContext[PlatformUserAuth], document: UploadedFile) -> str:
media_type = document.media_type.split(";", 1)[0].strip().lower()
if media_type not in ALLOWED_BROWSER_MEDIA_TYPES:
raise UploadRejected("unsupported_media_type", "Upload a UTF-8 text contract file.")
if document.size_bytes > MAX_UPLOAD_BYTES:
raise UploadRejected("upload_too_large", "Contract uploads are limited to 512 KB.")
try:
raw = ctx.workspace.read_bytes(document.path)
except Exception as exc: # noqa: BLE001
raise UploadRejected("upload_read_failed", "Could not read the uploaded contract file.") from exc
return _decode_uploaded_bytes(raw)
def _decode_uploaded_bytes(raw: bytes) -> str:
if not raw:
raise UploadRejected("empty_upload", "Uploaded contract file is empty.")
if len(raw) > MAX_UPLOAD_BYTES:
raise UploadRejected("upload_too_large", "Contract uploads are limited to 512 KB.")
try:
text = raw.decode("utf-8")
except UnicodeDecodeError as exc:
raise UploadRejected("invalid_text_encoding", "Uploaded contract must be UTF-8 text.") from exc
if not text.strip():
raise UploadRejected("empty_upload", "Uploaded contract file is empty.")
return text
async def _save_analysis(
*,
tenant_key: str,
contract_id: str,
title: str,
text: str,
analysis: dict[str, Any],
skill_name: str,
task_id: str,
) -> None:
await asyncio.to_thread(
_save_analysis_sync,
tenant_key=tenant_key,
contract_id=contract_id,
title=title,
text=text,
analysis=analysis,
skill_name=skill_name,
task_id=task_id,
)
async def _load_contract(*, tenant_key: str, contract_id: str) -> dict[str, Any] | None:
return await asyncio.to_thread(_load_contract_sync, tenant_key=tenant_key, contract_id=contract_id)
def _connect_db() -> Any:
dsn = os.environ.get("DATABASE_URL", "").strip()
if not dsn:
raise DatabaseUnavailable("database_unavailable", "DATABASE_URL is not configured")
try:
import psycopg
except Exception as exc: # noqa: BLE001
raise DatabaseUnavailable("database_driver_missing", "psycopg is not installed") from exc
try:
return psycopg.connect(
dsn,
connect_timeout=5,
options="-c statement_timeout=5000 -c idle_in_transaction_session_timeout=5000",
) )
except Exception as exc: # noqa: BLE001
raise DatabaseUnavailable("database_unavailable", "could not connect to managed Postgres") from exc
def _save_analysis_sync( def _runtime_skills_root(ctx: RunContext[Any]) -> str:
*, workspace = getattr(ctx, "_workspace", None)
tenant_key: str, prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
contract_id: str, if not prefixes:
title: str, outputs_prefix = getattr(workspace, "outputs_prefix", None)
text: str, prefixes = (outputs_prefix or "outputs/",)
analysis: dict[str, Any], prefix = str(prefixes[0]).strip("/")
skill_name: str, return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
task_id: str,
) -> None:
try:
from psycopg.types.json import Jsonb
except Exception as exc: # noqa: BLE001
raise DatabaseUnavailable("database_driver_missing", "psycopg JSON support is not installed") from exc
now = datetime.now(timezone.utc).isoformat()
input_hash = hashlib.sha256(
f"{tenant_key}\0{contract_id}\0{text}".encode("utf-8")
).hexdigest()
receipt = {
"agent": ContractClockStudioV1.name,
"version": ContractClockStudioV1.version,
"skill": skill_name,
"task_id": task_id,
"input_hash": input_hash,
"status": "ok",
"created_at": now,
}
analysis_with_receipt = {**analysis, "receipt": receipt}
try:
with _connect_db() as conn:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO contract_clock_contracts
(tenant_key, contract_id, title, source_text_sha256, deadlines, analysis, updated_at)
VALUES (%s, %s, %s, %s, %s, %s, NOW())
ON CONFLICT (tenant_key, contract_id) DO UPDATE SET
title = EXCLUDED.title,
source_text_sha256 = EXCLUDED.source_text_sha256,
deadlines = EXCLUDED.deadlines,
analysis = EXCLUDED.analysis,
updated_at = NOW()
""",
(
tenant_key,
contract_id,
title,
hashlib.sha256(text.encode("utf-8")).hexdigest(),
Jsonb(analysis["deadlines"]),
Jsonb(analysis_with_receipt),
),
)
cur.execute(
"""
INSERT INTO contract_clock_execution_receipts
(tenant_key, contract_id, skill_name, task_id, input_hash, status, result_summary)
VALUES (%s, %s, %s, %s, %s, %s, %s)
""",
(
tenant_key,
contract_id,
skill_name,
task_id,
input_hash,
"ok",
Jsonb(
{
"deadline_count": len(analysis.get("deadlines") or []),
"contract_id": contract_id,
}
),
),
)
except DatabaseUnavailable:
raise
except Exception as exc: # noqa: BLE001
raise DatabaseUnavailable("database_write_failed", "could not persist contract analysis") from exc
def _load_contract_sync(*, tenant_key: str, contract_id: str) -> dict[str, Any] | None: def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
try: """Copy packaged DeepAgents skills into the invocation workspace.
with _connect_db() as conn:
with conn.cursor() as cur: DeepAgents loads skills from its backend, while source-controlled
cur.execute( ``skills/`` folders live in the image. This bridge lets generated agents
""" ship reusable SKILL.md bundles without giving up durable A2A workspace
SELECT title, deadlines, analysis, updated_at files.
FROM contract_clock_contracts """
WHERE tenant_key = %s AND contract_id = %s root = Path(__file__).parent / "skills"
""", if not root.exists():
(tenant_key, contract_id), return []
) runtime_skills_root = _runtime_skills_root(ctx)
row = cur.fetchone() uploads: list[tuple[str, bytes]] = []
except DatabaseUnavailable: for path in root.rglob("*"):
raise if path.is_file():
except Exception as exc: # noqa: BLE001 rel = path.relative_to(root).as_posix()
raise DatabaseUnavailable("database_read_failed", "could not load contract timeline") from exc uploads.append((runtime_skills_root + rel, path.read_bytes()))
if row is None: if uploads:
return None backend.upload_files(uploads)
title, deadlines, analysis, updated_at = row return [runtime_skills_root]
analysis = analysis or {} return []
return {
"title": title,
"deadlines": deadlines or [], def _last_message_text(state: dict[str, Any]) -> str:
"receipt": analysis.get("receipt") or {}, messages = state.get("messages") or []
"updated_at": updated_at.isoformat() if hasattr(updated_at, "isoformat") else str(updated_at), if not messages:
} return json.dumps(state, default=str)
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])