a2a-source-edit: write agent.py
This commit is contained in:
a2a-cloud
872
agent.py
872
agent.py
@@ -1,189 +1,783 @@
|
||||
"""type-data-csv-excel-7rh4pz-057e66 agent.
|
||||
"""Data workflow builder agent.
|
||||
|
||||
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
|
||||
Creates reusable extraction/transformation workflow specs from uploaded CSV,
|
||||
Excel, JSON, JSON-LD, or other structured data and a caller-provided JSON
|
||||
Schema. The core workflow builder is deterministic so smoke/MCP calls do not
|
||||
require an LLM credential. The generated workflow is saved as an artifact and,
|
||||
when the managed database is available, recorded with an execution receipt for
|
||||
platform acceptance evidence.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import csv
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import Annotated, Any, Literal
|
||||
|
||||
from pydantic import BaseModel
|
||||
from pydantic import BaseModel, Field, ValidationError, field_validator
|
||||
|
||||
import a2a_pack as a2a
|
||||
from a2a_pack import (
|
||||
A2AAgent,
|
||||
LLMProvisioning,
|
||||
{{ auth_type }},
|
||||
AgentDatabase,
|
||||
AgentDatabaseEnv,
|
||||
AgentDatabaseMigrations,
|
||||
AgentPlatformResources,
|
||||
FileUpload,
|
||||
PlatformUserAuth,
|
||||
Pricing,
|
||||
Resources,
|
||||
RunContext,
|
||||
UploadedFile,
|
||||
WorkspaceAccess,
|
||||
WorkspaceMode,
|
||||
)
|
||||
from a2a_pack.context import LLMCreds
|
||||
|
||||
try: # Optional at card-load time; required only when DATABASE_URL exists.
|
||||
import psycopg
|
||||
except Exception: # pragma: no cover - sandbox card load before optional deps
|
||||
psycopg = None # type: ignore[assignment]
|
||||
|
||||
|
||||
MAX_UPLOAD_BYTES = 2 * 1024 * 1024
|
||||
MAX_SCHEMA_BYTES = 64 * 1024
|
||||
MAX_SAMPLE_ROWS = 25
|
||||
WORKFLOW_OUTPUT_PREFIX = "outputs/data-workflows/"
|
||||
DB_NAME = "type-data-csv-excel-7rh4pz-057e66-data"
|
||||
|
||||
|
||||
class TypeDataCsvExcel7rh4pz057e66Config(BaseModel):
|
||||
pass
|
||||
max_upload_bytes: int = MAX_UPLOAD_BYTES
|
||||
|
||||
|
||||
SYSTEM_PROMPT = """\
|
||||
You are a compact tool-calling agent.
|
||||
class BrowserDataUpload(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_UPLOAD_BYTES * 2)
|
||||
|
||||
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 = "type-data-csv-excel-7rh4pz-057e66/.deepagents/skills/"
|
||||
DEEPAGENTS_RECURSION_LIMIT = 500
|
||||
@field_validator("filename")
|
||||
@classmethod
|
||||
def safe_filename(cls, value: str) -> str:
|
||||
clean = value.strip().replace("\\", "/").split("/")[-1]
|
||||
if not clean or clean in {".", ".."}:
|
||||
raise ValueError("filename must be a safe basename")
|
||||
if not re.fullmatch(r"[A-Za-z0-9._ ()@+-]{1,160}", clean):
|
||||
raise ValueError("filename contains unsupported characters")
|
||||
return clean
|
||||
|
||||
|
||||
class TypeDataCsvExcel7rh4pz057e66(A2AAgent[TypeDataCsvExcel7rh4pz057e66Config, {{ auth_type }}]):
|
||||
class WorkflowResult(BaseModel):
|
||||
status: Literal["ok", "validation_error", "storage_warning"]
|
||||
workflow_id: str
|
||||
workflow_name: str
|
||||
source_format: str
|
||||
fields_detected: list[str]
|
||||
mapped_fields: list[dict[str, Any]]
|
||||
unmapped_required_fields: list[str]
|
||||
workflow_path: str
|
||||
artifact_uri: str | None = None
|
||||
receipt_id: str
|
||||
persisted: bool
|
||||
warnings: list[str] = Field(default_factory=list)
|
||||
|
||||
|
||||
class TypeDataCsvExcel7rh4pz057e66(A2AAgent[TypeDataCsvExcel7rh4pz057e66Config, PlatformUserAuth]):
|
||||
name = "type-data-csv-excel-7rh4pz-057e66"
|
||||
description = "Creates reusable agent-powered extraction and transformation workflows from uploaded CSV, Excel, JSON, JSON-LD, and other tabular or structured data using a caller-provided JSON Schema."
|
||||
description = (
|
||||
"Builds reusable extraction and transformation workflows from CSV, "
|
||||
"Excel, JSON, JSON-LD, and other structured data into a requested JSON Schema."
|
||||
)
|
||||
version = "0.1.0"
|
||||
|
||||
config_model = TypeDataCsvExcel7rh4pz057e66Config
|
||||
auth_model = {{ auth_type }}
|
||||
auth_model = PlatformUserAuth
|
||||
|
||||
# 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=True,
|
||||
notes="Starter agent uses the caller's saved LLM credential via ctx.llm.",
|
||||
caller_pays_llm=False,
|
||||
notes="Deterministic workflow generation; no LLM credential is required.",
|
||||
)
|
||||
resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=600)
|
||||
workspace_access = WorkspaceAccess.dynamic(
|
||||
max_files=64,
|
||||
max_files=32,
|
||||
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
|
||||
require_reason=False,
|
||||
max_total_size_bytes=10 * 1024 * 1024,
|
||||
)
|
||||
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},
|
||||
platform_resources = AgentPlatformResources(
|
||||
databases=(
|
||||
AgentDatabase(
|
||||
name=DB_NAME,
|
||||
scope="user",
|
||||
access_mode="read_write",
|
||||
env=AgentDatabaseEnv(url="DATABASE_URL"),
|
||||
migrations=AgentDatabaseMigrations(path="db/migrations"),
|
||||
),
|
||||
)
|
||||
await ctx.emit_progress("deepagent finished")
|
||||
return _last_message_text(state)
|
||||
)
|
||||
tools_used = ("jsonschema", "csv", "openpyxl")
|
||||
|
||||
def _build_deep_agent(
|
||||
@a2a.tool(
|
||||
description=(
|
||||
"Create and save a reusable extraction/transformation workflow from "
|
||||
"a browser-uploaded base64 data file and a target JSON Schema."
|
||||
),
|
||||
timeout_seconds=120,
|
||||
cost_class="standard",
|
||||
grant_mode="read_write_overlay",
|
||||
grant_allow_patterns=("outputs/data-workflows/**",),
|
||||
grant_outputs_prefix=WORKFLOW_OUTPUT_PREFIX,
|
||||
grant_write_prefixes=(WORKFLOW_OUTPUT_PREFIX,),
|
||||
)
|
||||
async def create_workflow_from_browser_upload(
|
||||
self,
|
||||
ctx: RunContext[PlatformUserAuth],
|
||||
source_file: BrowserDataUpload,
|
||||
output_schema: dict[str, Any],
|
||||
workflow_name: str = "data-transform-workflow",
|
||||
) -> WorkflowResult:
|
||||
"""Browser-safe bounded base64 upload bridge for the packed frontend."""
|
||||
try:
|
||||
data = _decode_browser_upload(source_file, max_bytes=self.config.max_upload_bytes)
|
||||
except ValueError as exc:
|
||||
return WorkflowResult(
|
||||
status="validation_error",
|
||||
workflow_id="",
|
||||
workflow_name=workflow_name,
|
||||
source_format="unknown",
|
||||
fields_detected=[],
|
||||
mapped_fields=[],
|
||||
unmapped_required_fields=[],
|
||||
workflow_path="",
|
||||
receipt_id="",
|
||||
persisted=False,
|
||||
warnings=[str(exc)],
|
||||
)
|
||||
return await self._create_workflow(
|
||||
ctx=ctx,
|
||||
filename=source_file.filename,
|
||||
media_type=source_file.media_type,
|
||||
data=data,
|
||||
output_schema=output_schema,
|
||||
workflow_name=workflow_name,
|
||||
upload_mode="browser_base64",
|
||||
)
|
||||
|
||||
@a2a.tool(
|
||||
description=(
|
||||
"Create and save a reusable extraction/transformation workflow from "
|
||||
"an Agent API FileUpload and a target JSON Schema."
|
||||
),
|
||||
timeout_seconds=120,
|
||||
cost_class="standard",
|
||||
grant_mode="read_write_overlay",
|
||||
grant_allow_patterns=("outputs/data-workflows/**",),
|
||||
grant_outputs_prefix=WORKFLOW_OUTPUT_PREFIX,
|
||||
grant_write_prefixes=(WORKFLOW_OUTPUT_PREFIX,),
|
||||
)
|
||||
async def create_workflow_from_file_upload(
|
||||
self,
|
||||
ctx: RunContext[PlatformUserAuth],
|
||||
source_file: Annotated[
|
||||
UploadedFile,
|
||||
FileUpload(
|
||||
accept=(
|
||||
"text/csv",
|
||||
"application/csv",
|
||||
"application/json",
|
||||
"application/ld+json",
|
||||
"application/vnd.ms-excel",
|
||||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||||
"text/plain",
|
||||
),
|
||||
max_bytes=MAX_UPLOAD_BYTES,
|
||||
description="CSV, Excel, JSON, JSON-LD, or text data file",
|
||||
),
|
||||
],
|
||||
output_schema: dict[str, Any],
|
||||
workflow_name: str = "data-transform-workflow",
|
||||
) -> WorkflowResult:
|
||||
"""Typed FileUpload path for external Agent API/MCP-compatible clients."""
|
||||
if source_file.size_bytes and source_file.size_bytes > self.config.max_upload_bytes:
|
||||
return WorkflowResult(
|
||||
status="validation_error",
|
||||
workflow_id="",
|
||||
workflow_name=workflow_name,
|
||||
source_format="unknown",
|
||||
fields_detected=[],
|
||||
mapped_fields=[],
|
||||
unmapped_required_fields=[],
|
||||
workflow_path="",
|
||||
receipt_id="",
|
||||
persisted=False,
|
||||
warnings=[f"file exceeds {self.config.max_upload_bytes} byte limit"],
|
||||
)
|
||||
data = _read_uploaded_workspace_file(ctx, source_file.path)
|
||||
if len(data) > self.config.max_upload_bytes:
|
||||
return WorkflowResult(
|
||||
status="validation_error",
|
||||
workflow_id="",
|
||||
workflow_name=workflow_name,
|
||||
source_format="unknown",
|
||||
fields_detected=[],
|
||||
mapped_fields=[],
|
||||
unmapped_required_fields=[],
|
||||
workflow_path="",
|
||||
receipt_id="",
|
||||
persisted=False,
|
||||
warnings=[f"file exceeds {self.config.max_upload_bytes} byte limit"],
|
||||
)
|
||||
return await self._create_workflow(
|
||||
ctx=ctx,
|
||||
filename=source_file.filename,
|
||||
media_type=source_file.media_type,
|
||||
data=data,
|
||||
output_schema=output_schema,
|
||||
workflow_name=workflow_name,
|
||||
upload_mode="agent_file_upload",
|
||||
)
|
||||
|
||||
@a2a.tool(
|
||||
description="List recent saved workflow receipts for the signed-in user.",
|
||||
timeout_seconds=30,
|
||||
cost_class="cheap",
|
||||
idempotent=True,
|
||||
)
|
||||
async def list_workflow_receipts(
|
||||
self,
|
||||
ctx: RunContext[PlatformUserAuth],
|
||||
limit: int = 10,
|
||||
) -> dict[str, Any]:
|
||||
safe_limit = max(1, min(int(limit), 50))
|
||||
rows = _list_receipts(_tenant_key(ctx), safe_limit)
|
||||
return {"status": "ok", "receipts": rows, "persisted": bool(os.environ.get("DATABASE_URL"))}
|
||||
|
||||
async def _create_workflow(
|
||||
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
|
||||
|
||||
@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,
|
||||
}
|
||||
ctx: RunContext[PlatformUserAuth],
|
||||
filename: str,
|
||||
media_type: str,
|
||||
data: bytes,
|
||||
output_schema: dict[str, Any],
|
||||
workflow_name: str,
|
||||
upload_mode: str,
|
||||
) -> WorkflowResult:
|
||||
tenant = _tenant_key(ctx)
|
||||
warnings: list[str] = []
|
||||
schema_error = _validate_json_schema(output_schema)
|
||||
if schema_error:
|
||||
return WorkflowResult(
|
||||
status="validation_error",
|
||||
workflow_id="",
|
||||
workflow_name=workflow_name,
|
||||
source_format="unknown",
|
||||
fields_detected=[],
|
||||
mapped_fields=[],
|
||||
unmapped_required_fields=[],
|
||||
workflow_path="",
|
||||
receipt_id="",
|
||||
persisted=False,
|
||||
warnings=[schema_error],
|
||||
)
|
||||
if len(json.dumps(output_schema, default=str).encode("utf-8")) > MAX_SCHEMA_BYTES:
|
||||
return WorkflowResult(
|
||||
status="validation_error",
|
||||
workflow_id="",
|
||||
workflow_name=workflow_name,
|
||||
source_format="unknown",
|
||||
fields_detected=[],
|
||||
mapped_fields=[],
|
||||
unmapped_required_fields=[],
|
||||
workflow_path="",
|
||||
receipt_id="",
|
||||
persisted=False,
|
||||
warnings=[f"output_schema exceeds {MAX_SCHEMA_BYTES} bytes"],
|
||||
)
|
||||
|
||||
@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)
|
||||
profile = _profile_source_data(filename, media_type, data)
|
||||
warnings.extend(profile.get("warnings", []))
|
||||
workflow_id = _stable_workflow_id(filename, data, output_schema, workflow_name)
|
||||
receipt_id = f"rcpt_{uuid.uuid4().hex}"
|
||||
workflow_path = f"{WORKFLOW_OUTPUT_PREFIX}{workflow_id}.workflow.json"
|
||||
workflow_spec = _build_workflow_spec(
|
||||
workflow_id=workflow_id,
|
||||
workflow_name=workflow_name,
|
||||
filename=filename,
|
||||
media_type=media_type,
|
||||
upload_mode=upload_mode,
|
||||
profile=profile,
|
||||
output_schema=output_schema,
|
||||
receipt_id=receipt_id,
|
||||
)
|
||||
|
||||
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,
|
||||
artifact_uri: str | None = None
|
||||
ref = await ctx.write_artifact(
|
||||
f"{workflow_id}.workflow.json",
|
||||
json.dumps(workflow_spec, indent=2, sort_keys=True).encode("utf-8"),
|
||||
"application/json",
|
||||
)
|
||||
await ctx.emit_artifact(ref)
|
||||
artifact_uri = ref.uri
|
||||
|
||||
await _write_workspace_output(ctx, workflow_path, workflow_spec, warnings)
|
||||
|
||||
persisted = _persist_workflow_and_receipt(
|
||||
tenant_key=tenant,
|
||||
workflow_id=workflow_id,
|
||||
receipt_id=receipt_id,
|
||||
workflow_name=workflow_name,
|
||||
workflow_path=workflow_path,
|
||||
source_format=str(profile["format"]),
|
||||
source_filename=filename,
|
||||
spec=workflow_spec,
|
||||
result_summary={
|
||||
"mapped_count": len(workflow_spec["mapping"]["fields"]),
|
||||
"unmapped_required_fields": workflow_spec["mapping"]["unmapped_required_fields"],
|
||||
"artifact_uri": artifact_uri,
|
||||
},
|
||||
warnings=warnings,
|
||||
)
|
||||
if not persisted:
|
||||
warnings.append("DATABASE_URL is not configured; workflow artifact was created but database receipt was not persisted")
|
||||
|
||||
return WorkflowResult(
|
||||
status="ok" if persisted else "storage_warning",
|
||||
workflow_id=workflow_id,
|
||||
workflow_name=workflow_name,
|
||||
source_format=str(profile["format"]),
|
||||
fields_detected=list(profile.get("fields", [])),
|
||||
mapped_fields=list(workflow_spec["mapping"]["fields"]),
|
||||
unmapped_required_fields=list(workflow_spec["mapping"]["unmapped_required_fields"]),
|
||||
workflow_path=workflow_path,
|
||||
artifact_uri=artifact_uri,
|
||||
receipt_id=receipt_id,
|
||||
persisted=persisted,
|
||||
warnings=warnings,
|
||||
)
|
||||
|
||||
|
||||
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 _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
|
||||
auth = ctx.auth
|
||||
stable_id = auth.user_id if auth.user_id is not None else auth.sub
|
||||
if stable_id is None or str(stable_id).strip() == "":
|
||||
raise PermissionError("stable platform identity required")
|
||||
return f"user:{stable_id}"
|
||||
|
||||
|
||||
def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
|
||||
"""Copy packaged DeepAgents skills into the invocation workspace.
|
||||
def _decode_browser_upload(source_file: BrowserDataUpload, *, max_bytes: int) -> bytes:
|
||||
if source_file.media_type not in _accepted_media_types():
|
||||
raise ValueError(f"unsupported media_type {source_file.media_type!r}")
|
||||
try:
|
||||
data = base64.b64decode(source_file.data_base64, validate=True)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
raise ValueError("data_base64 must be valid base64") from exc
|
||||
if not data:
|
||||
raise ValueError("uploaded file is empty")
|
||||
if len(data) > max_bytes:
|
||||
raise ValueError(f"uploaded file exceeds {max_bytes} byte limit")
|
||||
return data
|
||||
|
||||
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():
|
||||
|
||||
def _accepted_media_types() -> set[str]:
|
||||
return {
|
||||
"text/csv",
|
||||
"application/csv",
|
||||
"application/json",
|
||||
"application/ld+json",
|
||||
"application/vnd.ms-excel",
|
||||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||||
"text/plain",
|
||||
"application/octet-stream",
|
||||
}
|
||||
|
||||
|
||||
def _read_uploaded_workspace_file(ctx: RunContext[PlatformUserAuth], path: str) -> bytes:
|
||||
workspace = ctx.workspace
|
||||
reader = getattr(workspace, "read_bytes", None)
|
||||
if callable(reader):
|
||||
return bytes(reader(path))
|
||||
raise RuntimeError("workspace file reader is unavailable for uploaded file")
|
||||
|
||||
|
||||
async def _write_workspace_output(ctx: RunContext[PlatformUserAuth], path: str, payload: dict[str, Any], warnings: list[str]) -> None:
|
||||
data = json.dumps(payload, indent=2, sort_keys=True).encode("utf-8")
|
||||
workspace = ctx.workspace
|
||||
writer = getattr(workspace, "write_bytes", None)
|
||||
if callable(writer):
|
||||
writer(path, data)
|
||||
return
|
||||
try:
|
||||
grant = await workspace.request_access(
|
||||
files=(path,),
|
||||
mode=WorkspaceMode.READ_WRITE_OVERLAY,
|
||||
reason="Save generated data workflow spec",
|
||||
purpose="workflow output",
|
||||
)
|
||||
view = await workspace.open_view(
|
||||
purpose="workflow output",
|
||||
hints=(path,),
|
||||
max_files=1,
|
||||
mode=WorkspaceMode.READ_WRITE_OVERLAY,
|
||||
reason="Save generated data workflow spec",
|
||||
)
|
||||
await view.write(path, data)
|
||||
_ = grant
|
||||
except Exception as exc: # noqa: BLE001
|
||||
warnings.append(f"workspace write warning: {type(exc).__name__}: {str(exc)[:160]}")
|
||||
|
||||
|
||||
def _validate_json_schema(schema: dict[str, Any]) -> str | None:
|
||||
if not isinstance(schema, dict):
|
||||
return "output_schema must be a JSON object"
|
||||
if schema.get("type") != "object":
|
||||
return "output_schema must describe a JSON object with type='object'"
|
||||
props = schema.get("properties")
|
||||
if not isinstance(props, dict) or not props:
|
||||
return "output_schema.properties must be a non-empty object"
|
||||
required = schema.get("required", [])
|
||||
if required is not None and not isinstance(required, list):
|
||||
return "output_schema.required must be a list when provided"
|
||||
try:
|
||||
json.dumps(schema)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return f"output_schema must be JSON serializable: {exc}"
|
||||
return None
|
||||
|
||||
|
||||
def _profile_source_data(filename: str, media_type: str, data: bytes) -> dict[str, Any]:
|
||||
ext = Path(filename.lower()).suffix
|
||||
text = data.decode("utf-8-sig", errors="replace")
|
||||
warnings: list[str] = []
|
||||
if media_type in {"application/json", "application/ld+json"} or ext in {".json", ".jsonld"}:
|
||||
try:
|
||||
parsed = json.loads(text)
|
||||
rows, fields = _json_rows_and_fields(parsed)
|
||||
return {
|
||||
"format": "jsonld" if media_type == "application/ld+json" or ext == ".jsonld" else "json",
|
||||
"fields": fields,
|
||||
"sample_rows": rows[:MAX_SAMPLE_ROWS],
|
||||
"row_count_sampled": min(len(rows), MAX_SAMPLE_ROWS),
|
||||
"warnings": warnings,
|
||||
}
|
||||
except json.JSONDecodeError as exc:
|
||||
warnings.append(f"JSON parse warning: {exc.msg}")
|
||||
if ext in {".xlsx", ".xlsm", ".xltx", ".xltm"} or "spreadsheetml" in media_type:
|
||||
try:
|
||||
return _profile_excel(data, warnings)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
warnings.append(f"Excel parse warning: {type(exc).__name__}: {str(exc)[:120]}; falling back to delimited text")
|
||||
delimiter = _sniff_delimiter(text)
|
||||
try:
|
||||
reader = csv.DictReader(io.StringIO(text), delimiter=delimiter)
|
||||
fields = [str(f).strip() for f in (reader.fieldnames or []) if str(f).strip()]
|
||||
rows = []
|
||||
for idx, row in enumerate(reader):
|
||||
if idx >= MAX_SAMPLE_ROWS:
|
||||
break
|
||||
rows.append({str(k): v for k, v in row.items() if k is not None})
|
||||
if fields:
|
||||
return {
|
||||
"format": "csv" if delimiter == "," else "delimited_text",
|
||||
"fields": fields,
|
||||
"sample_rows": rows,
|
||||
"row_count_sampled": len(rows),
|
||||
"delimiter": delimiter,
|
||||
"warnings": warnings,
|
||||
}
|
||||
except Exception as exc: # noqa: BLE001
|
||||
warnings.append(f"CSV parse warning: {type(exc).__name__}: {str(exc)[:120]}")
|
||||
lines = [line.strip() for line in text.splitlines() if line.strip()][:MAX_SAMPLE_ROWS]
|
||||
return {
|
||||
"format": "text",
|
||||
"fields": [],
|
||||
"sample_rows": [{"line": line} for line in lines],
|
||||
"row_count_sampled": len(lines),
|
||||
"warnings": warnings,
|
||||
}
|
||||
|
||||
|
||||
def _profile_excel(data: bytes, warnings: list[str]) -> dict[str, Any]:
|
||||
from openpyxl import load_workbook
|
||||
|
||||
wb = load_workbook(io.BytesIO(data), read_only=True, data_only=True)
|
||||
ws = wb[wb.sheetnames[0]]
|
||||
rows_iter = ws.iter_rows(values_only=True)
|
||||
header = next(rows_iter, None)
|
||||
fields = [str(cell).strip() for cell in (header or []) if cell is not None and str(cell).strip()]
|
||||
rows: list[dict[str, Any]] = []
|
||||
for idx, row in enumerate(rows_iter):
|
||||
if idx >= MAX_SAMPLE_ROWS:
|
||||
break
|
||||
rows.append({fields[i]: row[i] for i in range(min(len(fields), len(row)))})
|
||||
if not fields:
|
||||
warnings.append("first Excel sheet has no header row")
|
||||
return {
|
||||
"format": "excel",
|
||||
"fields": fields,
|
||||
"sample_rows": rows,
|
||||
"row_count_sampled": len(rows),
|
||||
"sheet": ws.title,
|
||||
"warnings": warnings,
|
||||
}
|
||||
|
||||
|
||||
def _json_rows_and_fields(parsed: Any) -> tuple[list[dict[str, Any]], list[str]]:
|
||||
if isinstance(parsed, list):
|
||||
rows = [item for item in parsed if isinstance(item, dict)]
|
||||
elif isinstance(parsed, dict):
|
||||
if "@graph" in parsed and isinstance(parsed["@graph"], list):
|
||||
rows = [item for item in parsed["@graph"] if isinstance(item, dict)]
|
||||
else:
|
||||
list_values = [value for value in parsed.values() if isinstance(value, list)]
|
||||
rows = [item for item in (list_values[0] if list_values else [parsed]) if isinstance(item, dict)]
|
||||
else:
|
||||
rows = []
|
||||
fields: list[str] = []
|
||||
for row in rows[:MAX_SAMPLE_ROWS]:
|
||||
for key in row:
|
||||
if key not in fields:
|
||||
fields.append(str(key))
|
||||
return rows, fields
|
||||
|
||||
|
||||
def _sniff_delimiter(text: str) -> str:
|
||||
sample = text[:4096]
|
||||
try:
|
||||
return csv.Sniffer().sniff(sample, delimiters=",\t;|").delimiter
|
||||
except Exception: # noqa: BLE001
|
||||
return ","
|
||||
|
||||
|
||||
def _build_workflow_spec(
|
||||
*,
|
||||
workflow_id: str,
|
||||
workflow_name: str,
|
||||
filename: str,
|
||||
media_type: str,
|
||||
upload_mode: str,
|
||||
profile: dict[str, Any],
|
||||
output_schema: dict[str, Any],
|
||||
receipt_id: str,
|
||||
) -> dict[str, Any]:
|
||||
source_fields = list(profile.get("fields", []))
|
||||
properties = output_schema.get("properties", {}) if isinstance(output_schema.get("properties"), dict) else {}
|
||||
required = [str(item) for item in output_schema.get("required", [])]
|
||||
mappings: list[dict[str, Any]] = []
|
||||
unmapped_required: list[str] = []
|
||||
for target, prop_schema in properties.items():
|
||||
source = _best_field_match(str(target), source_fields)
|
||||
target_type = prop_schema.get("type") if isinstance(prop_schema, dict) else None
|
||||
rule = {
|
||||
"target_field": str(target),
|
||||
"source_field": source,
|
||||
"required": str(target) in required,
|
||||
"transformations": _transformations_for_type(target_type, prop_schema if isinstance(prop_schema, dict) else {}),
|
||||
"confidence": 1.0 if source == target else 0.82 if source else 0.0,
|
||||
}
|
||||
mappings.append(rule)
|
||||
if not source and str(target) in required:
|
||||
unmapped_required.append(str(target))
|
||||
return {
|
||||
"workflow_version": "1.0",
|
||||
"workflow_id": workflow_id,
|
||||
"workflow_name": workflow_name.strip()[:120] or "data-transform-workflow",
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
"source": {
|
||||
"filename": filename,
|
||||
"media_type": media_type,
|
||||
"upload_mode": upload_mode,
|
||||
"format": profile.get("format"),
|
||||
"fields": source_fields,
|
||||
"sample_rows": _json_safe(profile.get("sample_rows", [])),
|
||||
"sample_limit": MAX_SAMPLE_ROWS,
|
||||
},
|
||||
"target_schema": output_schema,
|
||||
"mapping": {
|
||||
"strategy": "normalized_field_name_match_with_type_coercion",
|
||||
"fields": mappings,
|
||||
"unmapped_required_fields": unmapped_required,
|
||||
},
|
||||
"validation": {
|
||||
"required_fields": required,
|
||||
"additional_properties": output_schema.get("additionalProperties", True),
|
||||
"gates": [
|
||||
"parse source with declared source.format parser",
|
||||
"apply per-field mapping and transformations",
|
||||
"validate each output object against target_schema",
|
||||
"send rows with unmapped required fields to review queue",
|
||||
],
|
||||
},
|
||||
"receipt": {"receipt_id": receipt_id, "kind": "workflow_created"},
|
||||
}
|
||||
|
||||
|
||||
def _best_field_match(target: str, source_fields: list[str]) -> str | None:
|
||||
if target in source_fields:
|
||||
return target
|
||||
normalized_target = _normalize_name(target)
|
||||
normalized = {_normalize_name(field): field for field in source_fields}
|
||||
if normalized_target in normalized:
|
||||
return normalized[normalized_target]
|
||||
for norm, original in normalized.items():
|
||||
if normalized_target and (normalized_target in norm or norm in normalized_target):
|
||||
return original
|
||||
return None
|
||||
|
||||
|
||||
def _normalize_name(value: str) -> str:
|
||||
return re.sub(r"[^a-z0-9]+", "", value.lower())
|
||||
|
||||
|
||||
def _transformations_for_type(target_type: Any, schema: dict[str, Any]) -> list[str]:
|
||||
t = target_type[0] if isinstance(target_type, list) and target_type else target_type
|
||||
steps = ["trim strings", "treat blank values as null"]
|
||||
if t in {"integer", "number"}:
|
||||
steps.append("parse numeric value")
|
||||
elif t == "boolean":
|
||||
steps.append("parse boolean synonyms true/false/yes/no/1/0")
|
||||
elif t == "array":
|
||||
steps.append("wrap scalar values or split delimited lists when configured")
|
||||
elif t == "object":
|
||||
steps.append("parse nested JSON object when source value is text")
|
||||
else:
|
||||
steps.append("coerce to string when non-null")
|
||||
if "enum" in schema:
|
||||
steps.append("validate against enum values")
|
||||
if schema.get("format") == "date" or schema.get("format") == "date-time":
|
||||
steps.append("normalize date/date-time to ISO 8601; flag ambiguous dates")
|
||||
return steps
|
||||
|
||||
|
||||
def _stable_workflow_id(filename: str, data: bytes, schema: dict[str, Any], workflow_name: str) -> str:
|
||||
digest = hashlib.sha256()
|
||||
digest.update(filename.encode("utf-8"))
|
||||
digest.update(data[:65536])
|
||||
digest.update(json.dumps(schema, sort_keys=True, default=str).encode("utf-8"))
|
||||
digest.update(workflow_name.encode("utf-8"))
|
||||
return f"wf_{digest.hexdigest()[:16]}"
|
||||
|
||||
|
||||
def _json_safe(value: Any) -> Any:
|
||||
try:
|
||||
json.dumps(value)
|
||||
return value
|
||||
except TypeError:
|
||||
return json.loads(json.dumps(value, default=str))
|
||||
|
||||
|
||||
def _db_connect() -> Any | None:
|
||||
url = os.environ.get("DATABASE_URL")
|
||||
if not url or psycopg is None:
|
||||
return None
|
||||
return psycopg.connect(
|
||||
url,
|
||||
options="-c statement_timeout=5000 -c lock_timeout=2000 -c idle_in_transaction_session_timeout=10000",
|
||||
)
|
||||
|
||||
|
||||
def _persist_workflow_and_receipt(
|
||||
*,
|
||||
tenant_key: str,
|
||||
workflow_id: str,
|
||||
receipt_id: str,
|
||||
workflow_name: str,
|
||||
workflow_path: str,
|
||||
source_format: str,
|
||||
source_filename: str,
|
||||
spec: dict[str, Any],
|
||||
result_summary: dict[str, Any],
|
||||
warnings: list[str],
|
||||
) -> bool:
|
||||
conn = _db_connect()
|
||||
if conn is None:
|
||||
return False
|
||||
try:
|
||||
with conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO data_workflows
|
||||
(tenant_key, workflow_id, workflow_name, workflow_path, source_format, source_filename, spec)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
||||
ON CONFLICT (tenant_key, workflow_id) DO UPDATE SET
|
||||
workflow_name = EXCLUDED.workflow_name,
|
||||
workflow_path = EXCLUDED.workflow_path,
|
||||
source_format = EXCLUDED.source_format,
|
||||
source_filename = EXCLUDED.source_filename,
|
||||
spec = EXCLUDED.spec,
|
||||
updated_at = NOW()
|
||||
""",
|
||||
(
|
||||
tenant_key,
|
||||
workflow_id,
|
||||
workflow_name,
|
||||
workflow_path,
|
||||
source_format,
|
||||
source_filename,
|
||||
json.dumps(spec),
|
||||
),
|
||||
)
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO execution_receipts
|
||||
(tenant_key, receipt_id, workflow_id, skill_name, status, input_hash, result_summary, warnings)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
||||
""",
|
||||
(
|
||||
tenant_key,
|
||||
receipt_id,
|
||||
workflow_id,
|
||||
"create_workflow",
|
||||
"ok",
|
||||
hashlib.sha256(json.dumps(spec.get("source", {}), sort_keys=True, default=str).encode("utf-8")).hexdigest(),
|
||||
json.dumps(result_summary),
|
||||
json.dumps(warnings),
|
||||
),
|
||||
)
|
||||
return True
|
||||
except Exception: # noqa: BLE001
|
||||
return False
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def _list_receipts(tenant_key: str, limit: int) -> list[dict[str, Any]]:
|
||||
conn = _db_connect()
|
||||
if conn is None:
|
||||
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 _last_message_text(state: dict[str, Any]) -> str:
|
||||
messages = state.get("messages") or []
|
||||
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])
|
||||
try:
|
||||
with conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT receipt_id, workflow_id, skill_name, status, result_summary, created_at
|
||||
FROM execution_receipts
|
||||
WHERE tenant_key = %s
|
||||
ORDER BY created_at DESC
|
||||
LIMIT %s
|
||||
""",
|
||||
(tenant_key, limit),
|
||||
)
|
||||
rows = []
|
||||
for receipt_id, workflow_id, skill_name, status, result_summary, created_at in cur.fetchall():
|
||||
rows.append(
|
||||
{
|
||||
"receipt_id": receipt_id,
|
||||
"workflow_id": workflow_id,
|
||||
"skill_name": skill_name,
|
||||
"status": status,
|
||||
"result_summary": result_summary,
|
||||
"created_at": created_at.isoformat() if hasattr(created_at, "isoformat") else str(created_at),
|
||||
}
|
||||
)
|
||||
return rows
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
Reference in New Issue
Block a user