"""Data workflow builder agent. 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 Annotated, Any, Literal from pydantic import BaseModel, Field, ValidationError, field_validator import a2a_pack as a2a from a2a_pack import ( A2AAgent, AgentDatabase, AgentDatabaseEnv, AgentDatabaseMigrations, AgentPlatformResources, FileUpload, PlatformUserAuth, Pricing, Resources, RunContext, UploadedFile, WorkspaceAccess, WorkspaceMode, ) 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): max_upload_bytes: int = MAX_UPLOAD_BYTES 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) @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 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 = ( "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 = PlatformUserAuth pricing = Pricing( price_per_call_usd=0.0, 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=32, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=False, max_total_size_bytes=10 * 1024 * 1024, ) platform_resources = AgentPlatformResources( databases=( AgentDatabase( name=DB_NAME, scope="user", access_mode="read_write", env=AgentDatabaseEnv(url="DATABASE_URL"), migrations=AgentDatabaseMigrations(path="db/migrations"), ), ) ) tools_used = ("jsonschema", "csv", "openpyxl") @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[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"], ) 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, ) 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 _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 _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 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") try: workspace = ctx.workspace except PermissionError as exc: warnings.append(f"workspace write skipped: {str(exc)[:160]}") return 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 [] 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()