diff --git a/agent.py b/agent.py index 2aeb8d1..c8b852e 100644 --- a/agent.py +++ b/agent.py @@ -1,1186 +1,189 @@ -"""Production-safe deterministic customer integration engineering agent. +"""customer-integration-engineer agent. -This agent guides an integration from requirements through generated workspace -artifacts and sandbox-style acceptance evidence. It deliberately does not call -an LLM, read provider credentials, execute arbitrary shell, deploy, merge, or -mutate customer systems. All customer credentials stay behind caller-managed -secret references. +Starter stack: + - DeepAgents for tool-calling orchestration + - Caller-provided LLM credentials via ctx.llm + - A tiny model-call middleware hook you can replace with tracing, + routing, rate limits, or policy checks """ from __future__ import annotations -import asyncio -import hashlib -import ipaddress import json -import re -import socket -import time -from datetime import UTC, datetime -from typing import Any, Literal -from urllib.parse import urlparse +from pathlib import Path +from typing import Any + +from pydantic import BaseModel import a2a_pack as a2a from a2a_pack import ( A2AAgent, - ConsumerSetup, - ConsumerSetupField, - EgressPolicy, - NoAuth, + LLMProvisioning, + {{ auth_type }}, Pricing, - Resources, RunContext, WorkspaceAccess, WorkspaceMode, ) -from a2a_pack.context import AgentEvent -from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator +from a2a_pack.context import LLMCreds -MAX_JSON_BYTES = 256_000 -MAX_TEXT_BYTES = 128_000 -APPROVAL_PHRASE = "I_APPROVE_THE_LISTED_STEPS" -OUTPUT_ROOT = "outputs/integrations" -SECRET_WORDS = ("secret", "token", "password", "passwd", "apikey", "api_key", "authorization", "bearer") +class CustomerIntegrationEngineerConfig(BaseModel): + pass -def _tool_input_schema(request_model: type[BaseModel]) -> dict[str, Any]: - return { - "type": "object", - "required": ["request"], - "additionalProperties": False, - "properties": {"request": request_model.model_json_schema()}, - } +SYSTEM_PROMPT = """\ +You are a compact tool-calling agent. +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. +""" -class StrictModel(BaseModel): - model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) +RUNTIME_SKILLS_DIR = "customer-integration-engineer/.deepagents/skills/" +DEEPAGENTS_RECURSION_LIMIT = 500 -class AuditEntry(StrictModel): - ts: str - action: str - integration_id: str - details: dict[str, Any] = Field(default_factory=dict) - - -class ArtifactRecord(StrictModel): - path: str - sha256: str - size_bytes: int = Field(ge=0) - mime_type: str - - -class SkillStatus(StrictModel): - status: Literal["ok", "needs_setup", "blocked", "failed"] - integration_id: str - dry_run: bool - audit: list[AuditEntry] = Field(default_factory=list) - warnings: list[str] = Field(default_factory=list) - residual_risks: list[str] = Field(default_factory=list) - - -class SystemSpec(StrictModel): - name: str = Field(min_length=1, max_length=80) - kind: Literal[ - "rest_api", - "webhook", - "database", - "object_storage", - "observability", - "gitea", - "kubernetes", - "argo_cd", - "queue", - "other", - ] - base_url: str | None = Field(default=None, max_length=2048) - environment: str = Field(default="sandbox", max_length=64) - owner: str | None = Field(default=None, max_length=120) - notes: str | None = Field(default=None, max_length=2000) - - @field_validator("base_url") - @classmethod - def _safe_url_text(cls, value: str | None) -> str | None: - if value is None or not value: - return None - parsed = urlparse(value) - if parsed.scheme not in {"https", "http"} or not parsed.netloc: - raise ValueError("base_url must be an absolute http(s) URL") - return value - - -class AuthMethod(StrictModel): - system_name: str = Field(min_length=1, max_length=80) - auth_type: Literal["none", "api_key", "bearer_token", "oauth2_client_credentials", "hmac_signature", "mtls", "basic", "oidc"] - secret_ref: str | None = Field( - default=None, - pattern=r"^[A-Z][A-Z0-9_]{2,127}$", - description="Caller-managed secret reference name only; never a raw secret value.", - ) - header_name: str | None = Field(default=None, max_length=80) - token_url: str | None = Field(default=None, max_length=2048) - scopes: list[str] = Field(default_factory=list, max_length=20) - - @model_validator(mode="after") - def _require_secret_ref_for_secret_auth(self) -> "AuthMethod": - if self.auth_type != "none" and not self.secret_ref: - raise ValueError("secret_ref is required for non-none auth methods") - _reject_raw_secret(self.secret_ref or "") - return self - - -class FieldMapping(StrictModel): - source: str = Field(min_length=1, max_length=160) - target: str = Field(min_length=1, max_length=160) - transform: str = Field(default="copy", max_length=500) - required: bool = True - pii: bool = False - - -class DataContract(StrictModel): - name: str = Field(min_length=1, max_length=120) - direction: Literal["inbound", "outbound", "bidirectional"] - content_type: str = Field(default="application/json", max_length=120) - schema_ref: str | None = Field(default=None, max_length=500) - fields: list[FieldMapping] = Field(default_factory=list, max_length=200) - example_payload: dict[str, Any] | None = None - - @field_validator("example_payload") - @classmethod - def _bounded_example(cls, value: dict[str, Any] | None) -> dict[str, Any] | None: - if value is not None: - _ensure_bounded_json(value, MAX_JSON_BYTES) - return value - - -class RateLimitSpec(StrictModel): - system_name: str = Field(min_length=1, max_length=80) - max_requests: int = Field(ge=1, le=1_000_000) - per_seconds: int = Field(ge=1, le=86_400) - burst: int | None = Field(default=None, ge=1, le=1_000_000) - retry_after_header: str | None = Field(default="Retry-After", max_length=80) - - -class EnvironmentSpec(StrictModel): - name: str = Field(min_length=1, max_length=64) - purpose: Literal["dev", "test", "sandbox", "staging", "prod"] - endpoint_refs: list[str] = Field(default_factory=list, max_length=20) - object_storage_ref: str | None = Field(default=None, max_length=128) - observability_ref: str | None = Field(default=None, max_length=128) - mutable: bool = False - - -class ComplianceConstraint(StrictModel): - name: str = Field(min_length=1, max_length=120) - category: Literal["pii", "phi", "pci", "soc2", "gdpr", "hipaa", "data_residency", "retention", "audit", "other"] - requirement: str = Field(min_length=1, max_length=2000) - - -class SuccessCriterion(StrictModel): - name: str = Field(min_length=1, max_length=120) - measurement: str = Field(min_length=1, max_length=500) - target: str = Field(min_length=1, max_length=200) - - -class IntegrationRequirements(StrictModel): - integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$") - title: str = Field(min_length=1, max_length=160) - systems: list[SystemSpec] = Field(min_length=1, max_length=12) - auth_methods: list[AuthMethod] = Field(default_factory=list, max_length=12) - data_contracts: list[DataContract] = Field(default_factory=list, max_length=20) - rate_limits: list[RateLimitSpec] = Field(default_factory=list, max_length=20) - environments: list[EnvironmentSpec] = Field(default_factory=list, max_length=12) - compliance_constraints: list[ComplianceConstraint] = Field(default_factory=list, max_length=30) - success_criteria: list[SuccessCriterion] = Field(min_length=1, max_length=20) - external_host_allowlist: list[str] = Field(default_factory=list, max_length=50) - repository_write_allowlist: list[str] = Field(default_factory=list, max_length=20) - notes: str | None = Field(default=None, max_length=4000) - dry_run: bool = True - - @field_validator("external_host_allowlist") - @classmethod - def _clean_hosts(cls, value: list[str]) -> list[str]: - return [_validate_hostname(host) for host in value] - - @field_validator("repository_write_allowlist") - @classmethod - def _clean_repos(cls, value: list[str]) -> list[str]: - out: list[str] = [] - for repo in value: - clean = repo.strip() - if not re.fullmatch(r"[A-Za-z0-9_.-]{1,80}/[A-Za-z0-9_.-]{1,120}", clean): - raise ValueError("repository allowlist entries must look like owner/repo") - out.append(clean) - return out - - -class DesignIntegrationRequest(StrictModel): - requirements: IntegrationRequirements - create_artifacts: bool = True - dry_run: bool = True - - -class DesignIntegrationResult(SkillStatus): - architecture_path: str | None = None - mapping_path: str | None = None - plan_digest: str - artifacts: list[ArtifactRecord] = Field(default_factory=list) - architecture_summary: str - field_mapping_count: int = Field(ge=0) - required_setup: list[str] = Field(default_factory=list) - - -class GenerateConnectorRequest(StrictModel): - requirements: IntegrationRequirements - connector_language: Literal["python", "typescript"] = "python" - include_tests: bool = True - dry_run: bool = True - external_host_allowlist: list[str] = Field(default_factory=list, max_length=50) - repository_write_allowlist: list[str] = Field(default_factory=list, max_length=20) - - @field_validator("external_host_allowlist") - @classmethod - def _clean_hosts(cls, value: list[str]) -> list[str]: - return [_validate_hostname(host) for host in value] - - -class GenerateConnectorResult(SkillStatus): - connector_path: str | None = None - test_path: str | None = None - plan_digest: str - artifacts: list[ArtifactRecord] = Field(default_factory=list) - generated_files: list[str] = Field(default_factory=list) - blocked_actions: list[str] = Field(default_factory=list) - - -class ContractFixture(StrictModel): - name: str = Field(min_length=1, max_length=120) - payload: dict[str, Any] - should_pass: bool = True - - @field_validator("payload") - @classmethod - def _bounded_payload(cls, value: dict[str, Any]) -> Any: - _ensure_bounded_json(value, MAX_JSON_BYTES) - return value - - -class ContractArtifact(StrictModel): - name: str = Field(min_length=1, max_length=120) - kind: Literal["json_schema", "openapi_3", "webhook"] - schema_doc: dict[str, Any] - fixtures: list[ContractFixture] = Field(default_factory=list, max_length=50) - signature_header: str | None = Field(default=None, max_length=120) - secret_ref: str | None = Field(default=None, pattern=r"^[A-Z][A-Z0-9_]{2,127}$") - - @field_validator("schema_doc") - @classmethod - def _bounded_schema(cls, value: dict[str, Any]) -> Any: - _ensure_bounded_json(value, MAX_JSON_BYTES) - return value - - -class ValidateContractRequest(StrictModel): - integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$") - contracts: list[ContractArtifact] = Field(min_length=1, max_length=20) - fail_on_drift: bool = True - dry_run: bool = True - - -class ContractCheck(StrictModel): - name: str - kind: str - status: Literal["pass", "fail", "warn"] - findings: list[str] = Field(default_factory=list) - fixture_results: list[dict[str, Any]] = Field(default_factory=list) - - -class ValidateContractResult(SkillStatus): - contract_report_path: str | None = None - artifacts: list[ArtifactRecord] = Field(default_factory=list) - checks: list[ContractCheck] = Field(default_factory=list) - drift_detected: bool = False - - -class EndpointTest(StrictModel): - name: str = Field(min_length=1, max_length=120) - method: Literal["GET", "POST", "PUT", "PATCH", "DELETE"] = "POST" - url: str = Field(max_length=2048) - expected_status: int = Field(default=200, ge=100, le=599) - request_json: dict[str, Any] | None = None - required_secret_refs: list[str] = Field(default_factory=list, max_length=10) - idempotency_key: str | None = Field(default=None, max_length=160) - - @field_validator("request_json") - @classmethod - def _bounded_request(cls, value: dict[str, Any] | None) -> dict[str, Any] | None: - if value is not None: - _ensure_bounded_json(value, MAX_JSON_BYTES) - return value - - @field_validator("required_secret_refs") - @classmethod - def _secret_refs_only(cls, value: list[str]) -> list[str]: - for ref in value: - if not re.fullmatch(r"[A-Z][A-Z0-9_]{2,127}", ref): - raise ValueError("secret refs must be uppercase reference names") - _reject_raw_secret(ref) - return value - - -class RunAcceptanceRequest(StrictModel): - integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$") - endpoint_tests: list[EndpointTest] = Field(default_factory=list, max_length=20) - contract_fixtures: list[ContractArtifact] = Field(default_factory=list, max_length=20) - external_host_allowlist: list[str] = Field(default_factory=list, max_length=50) - allow_network: bool = False - max_retries: int = Field(default=1, ge=0, le=3) - timeout_seconds: float = Field(default=8.0, ge=1.0, le=30.0) - dry_run: bool = True - - @field_validator("external_host_allowlist") - @classmethod - def _clean_hosts(cls, value: list[str]) -> list[str]: - return [_validate_hostname(host) for host in value] - - -class AcceptanceStepResult(StrictModel): - name: str - status: Literal["pass", "fail", "skipped", "blocked"] - attempts: int = Field(ge=0) - detail: str - idempotency_key: str | None = None - - -class RunAcceptanceResult(SkillStatus): - acceptance_report_path: str | None = None - artifacts: list[ArtifactRecord] = Field(default_factory=list) - step_results: list[AcceptanceStepResult] = Field(default_factory=list) - contract_results: list[ContractCheck] = Field(default_factory=list) - network_calls_made: int = Field(ge=0) - - -class HandoffSection(StrictModel): - heading: str = Field(min_length=1, max_length=120) - body: str = Field(min_length=1, max_length=8000) - - @field_validator("body") - @classmethod - def _redact_body(cls, value: str) -> str: - return redact_secrets(value) - - -class PrepareHandoffRequest(StrictModel): - integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$") - title: str = Field(min_length=1, max_length=160) - rollout_steps: list[str] = Field(min_length=1, max_length=30) - rollback_steps: list[str] = Field(min_length=1, max_length=30) - monitoring_checks: list[str] = Field(min_length=1, max_length=30) - operator_notes: list[HandoffSection] = Field(default_factory=list, max_length=20) - delivery_actions: list[str] = Field(default_factory=list, max_length=20) - plan_digest: str | None = Field(default=None, max_length=128) - approval_phrase: str | None = Field(default=None, max_length=64) - approve_delivery: bool = False - dry_run: bool = True - - -class PrepareHandoffResult(SkillStatus): - handoff_path: str | None = None - plan_digest: str - artifacts: list[ArtifactRecord] = Field(default_factory=list) - approval_required: bool - approval_valid: bool - delivery_executed: bool = False - - -class CustomerIntegrationEngineerConfig(StrictModel): - default_dry_run: bool = True - - -_DESIGN_OUTPUT_SCHEMA = DesignIntegrationResult.model_json_schema() -_GENERATE_OUTPUT_SCHEMA = GenerateConnectorResult.model_json_schema() -_VALIDATE_OUTPUT_SCHEMA = ValidateContractResult.model_json_schema() -_ACCEPTANCE_OUTPUT_SCHEMA = RunAcceptanceResult.model_json_schema() -_HANDOFF_OUTPUT_SCHEMA = PrepareHandoffResult.model_json_schema() - - -class CustomerIntegrationEngineer(A2AAgent[CustomerIntegrationEngineerConfig, NoAuth]): +class CustomerIntegrationEngineer(A2AAgent[CustomerIntegrationEngineerConfig, {{ auth_type }}]): name = "customer-integration-engineer" - description = ( - "Guides and validates customer integrations from requirements through a tested handoff while " - "keeping credentials and production changes under customer control." - ) - version = "0.1.6" + description = "Validates the customer-integration-engineer release end to end, preserving existing source unless critical defects remain." + version = "0.1.0" config_model = CustomerIntegrationEngineerConfig - auth_model = NoAuth + auth_model = {{ auth_type }} + + # Hosted generated agents read the caller's saved LLM credential through + # ctx.llm. The platform may proxy that credential through LiteLLM, but agent + # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY + # directly. + llm_provisioning = LLMProvisioning.PLATFORM pricing = Pricing( price_per_call_usd=0.0, - caller_pays_llm=False, - notes="Deterministic no-LLM integration engineering; callers keep credentials and production changes under their control.", + caller_pays_llm=True, + notes="Starter agent uses the caller's saved LLM credential via ctx.llm.", ) - resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=600) - egress = EgressPolicy(deny_internet_by_default=True) workspace_access = WorkspaceAccess.dynamic( - max_files=256, + max_files=64, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), - require_reason=True, - deny_patterns=("**/.env", "**/*secret*", "**/*token*", "**/id_rsa", "**/.git/**"), - max_total_size_bytes=20 * 1024 * 1024, - ) - tools_used = ("pydantic", "jsonschema", "httpx") - consumer_setup = ConsumerSetup.from_fields( - ConsumerSetupField.config("GITEA_REPOSITORY", label="Gitea repository", description="Optional owner/repo target controlled by the caller.", required=False), - ConsumerSetupField.secret("GITEA_TOKEN", label="Gitea token", description="Optional caller-managed token reference; never returned or stored.", required=False), - ConsumerSetupField.config("API_BASE_URL", label="API base URL", description="Optional endpoint used only when allowlisted and dry_run=false.", required=False, input_type="url"), - ConsumerSetupField.secret("API_AUTH_TOKEN", label="API auth token", description="Optional caller-managed API credential reference.", required=False), - ConsumerSetupField.config("OBJECT_STORAGE_REF", label="Object storage reference", description="Caller-owned object storage config reference.", required=False), - ConsumerSetupField.config("OBSERVABILITY_REF", label="Observability reference", description="Caller-owned logs/metrics/traces reference.", required=False), - ConsumerSetupField.config("KUBERNETES_CONTEXT_REF", label="Kubernetes context reference", description="Optional caller-owned Kubernetes context reference.", required=False), - ConsumerSetupField.config("ARGOCD_APP_REF", label="Argo CD app reference", description="Optional caller-owned Argo CD app reference.", required=False), + require_reason=False, ) + tools_used = ("deepagents", "langchain") - @a2a.tool(description="Capture requirements, produce architecture and field mapping artifacts", input_schema=_tool_input_schema(DesignIntegrationRequest), timeout_seconds=120, idempotent=True, max_retries=0, cost_class="deterministic") - async def design_integration(self, ctx: RunContext[NoAuth], request: DesignIntegrationRequest) -> DesignIntegrationResult: - req = request.requirements.model_copy( - update={"dry_run": _effective_dry_run(self, request.dry_run, request.requirements.dry_run)} - ) - audit = [_audit("design_started", req.integration_id, {"dry_run": req.dry_run})] - await _emit(ctx, audit[-1]) - warnings = _requirements_warnings(req) - required_setup = _required_setup(req) - digest = _plan_digest({"phase": "design", "requirements": req.model_dump(mode="json")}) - architecture = _render_architecture(req, digest, warnings, required_setup) - mapping = _build_mapping(req) - artifacts: list[ArtifactRecord] = [] - arch_path = mapping_path = None - if request.create_artifacts: - base = _integration_base(req.integration_id) - arch_path = f"{base}/architecture.md" - mapping_path = f"{base}/mapping.json" - artifacts.append(await _write_workspace_text(ctx, arch_path, architecture, "text/markdown")) - artifacts.append(await _write_workspace_json(ctx, mapping_path, mapping)) - audit.append(_audit("design_completed", req.integration_id, {"artifacts": [a.path for a in artifacts]})) - await _emit(ctx, audit[-1]) - return DesignIntegrationResult( - status="ok", - integration_id=req.integration_id, - dry_run=req.dry_run, - audit=audit, - warnings=warnings, - residual_risks=_residual_risks(req), - architecture_path=arch_path, - mapping_path=mapping_path, - plan_digest=digest, - artifacts=artifacts, - architecture_summary=f"{req.title}: {len(req.systems)} systems, {len(req.data_contracts)} contracts, {len(req.success_criteria)} success criteria.", - field_mapping_count=sum(len(c.fields) for c in req.data_contracts), - required_setup=required_setup, + @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}, ) + await ctx.emit_progress("deepagent finished") + return _last_message_text(state) - @a2a.tool(description="Generate a minimal workspace-contained connector scaffold with no deployment or repository mutation", input_schema=_tool_input_schema(GenerateConnectorRequest), timeout_seconds=120, idempotent=True, max_retries=0, cost_class="deterministic") - async def generate_connector(self, ctx: RunContext[NoAuth], request: GenerateConnectorRequest) -> GenerateConnectorResult: - req = request.requirements.model_copy( - update={"dry_run": _effective_dry_run(self, request.dry_run, request.requirements.dry_run)} - ) - audit = [_audit("connector_generation_started", req.integration_id, {"language": request.connector_language, "dry_run": req.dry_run})] - await _emit(ctx, audit[-1]) - warnings = _requirements_warnings(req) - blocked_actions: list[str] = [] - if request.repository_write_allowlist or req.repository_write_allowlist: - blocked_actions.append("repository writes are never performed; generated source is written only under outputs/integrations/{integration_id}/") - allowlist = sorted(set(req.external_host_allowlist + request.external_host_allowlist)) - digest = _plan_digest({"phase": "generate_connector", "requirements": req.model_dump(mode="json"), "language": request.connector_language}) - base = _integration_base(req.integration_id) - artifacts: list[ArtifactRecord] = [] - generated_files: list[str] = [] - if request.connector_language == "python": - connector_path = f"{base}/connector/python/connector.py" - test_path = f"{base}/connector/python/test_connector.py" if request.include_tests else None - source = _python_connector(req, allowlist, digest) - artifacts.append(await _write_workspace_text(ctx, connector_path, source, "text/x-python")) - generated_files.append(connector_path) - if test_path: - test_source = _python_connector_test(req) - artifacts.append(await _write_workspace_text(ctx, test_path, test_source, "text/x-python")) - generated_files.append(test_path) - else: - connector_path = f"{base}/connector/typescript/connector.ts" - test_path = f"{base}/connector/typescript/connector.test.ts" if request.include_tests else None - source = _typescript_connector(req, allowlist, digest) - artifacts.append(await _write_workspace_text(ctx, connector_path, source, "text/typescript")) - generated_files.append(connector_path) - if test_path: - test_source = _typescript_connector_test(req) - artifacts.append(await _write_workspace_text(ctx, test_path, test_source, "text/typescript")) - generated_files.append(test_path) - audit.append(_audit("connector_generation_completed", req.integration_id, {"files": generated_files, "blocked_actions": blocked_actions})) - await _emit(ctx, audit[-1]) - status: Literal["ok", "needs_setup", "blocked", "failed"] = "blocked" if blocked_actions and not req.dry_run else "ok" - return GenerateConnectorResult( - status=status, - integration_id=req.integration_id, - dry_run=req.dry_run, - audit=audit, - warnings=warnings, - residual_risks=_residual_risks(req), - connector_path=connector_path, - test_path=test_path, - plan_digest=digest, - artifacts=artifacts, - generated_files=generated_files, - blocked_actions=blocked_actions, - ) + 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 - @a2a.tool(description="Validate OpenAPI, JSON Schema, and webhook contracts with deterministic fixtures", input_schema=_tool_input_schema(ValidateContractRequest), timeout_seconds=120, idempotent=True, max_retries=0, cost_class="deterministic") - async def validate_contract(self, ctx: RunContext[NoAuth], request: ValidateContractRequest) -> ValidateContractResult: - dry_run = _effective_dry_run(self, request.dry_run) - audit = [_audit("contract_validation_started", request.integration_id, {"contracts": len(request.contracts), "dry_run": dry_run})] - await _emit(ctx, audit[-1]) - checks = [_validate_one_contract(contract) for contract in request.contracts] - drift = any(check.status == "fail" for check in checks) - report = { - "integration_id": request.integration_id, - "status": "fail" if drift and request.fail_on_drift else "pass", - "dry_run": dry_run, - "generated_at": _now(), - "checks": [c.model_dump(mode="json") for c in checks], - "secret_handling": "secret_ref values only; raw credentials are rejected and redacted", - } - path = f"{_integration_base(request.integration_id)}/contract-report.json" - artifact = await _write_workspace_json(ctx, path, report) - audit.append(_audit("contract_validation_completed", request.integration_id, {"drift_detected": drift, "report": path})) - await _emit(ctx, audit[-1]) - return ValidateContractResult( - status="failed" if drift and request.fail_on_drift else "ok", - integration_id=request.integration_id, - dry_run=dry_run, - audit=audit, - warnings=[] if not drift else ["contract drift detected; inspect contract-report.json before handoff"], - residual_risks=[] if not drift else ["fixtures are deterministic and bounded; live provider behavior still requires caller-controlled staging verification"], - contract_report_path=path, - artifacts=[artifact], - checks=checks, - drift_detected=drift, - ) + @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, + } + ) - @a2a.tool(description="Run dry-run or allowlisted acceptance checks using endpoint metadata and secret references", input_schema=_tool_input_schema(RunAcceptanceRequest), timeout_seconds=180, idempotent=True, max_retries=0, cost_class="deterministic") - async def run_acceptance(self, ctx: RunContext[NoAuth], request: RunAcceptanceRequest) -> RunAcceptanceResult: - dry_run = _effective_dry_run(self, request.dry_run) - live_network_block = _live_network_unavailable_reason(self) - audit = [_audit("acceptance_started", request.integration_id, {"endpoint_tests": len(request.endpoint_tests), "dry_run": dry_run, "allow_network": request.allow_network})] - await _emit(ctx, audit[-1]) - contract_results = [_validate_one_contract(contract) for contract in request.contract_fixtures] - step_results: list[AcceptanceStepResult] = [] - network_calls = 0 - warnings: list[str] = [] - for test in request.endpoint_tests: - idem = test.idempotency_key or _plan_digest({"integration_id": request.integration_id, "test": test.model_dump(mode="json")})[:32] - if dry_run or not request.allow_network: - validation = _validate_endpoint_metadata(test.url, request.external_host_allowlist) - if validation: - step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=validation, idempotency_key=idem)) - warnings.append(validation) - else: - step_results.append(AcceptanceStepResult(name=test.name, status="skipped", attempts=0, detail="dry_run/default safe mode: endpoint metadata was validated but no DNS lookup or external request was sent", idempotency_key=idem)) - continue - metadata_validation = _validate_endpoint_metadata(test.url, request.external_host_allowlist) - if metadata_validation: - step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=metadata_validation, idempotency_key=idem)) - warnings.append(metadata_validation) - continue - if live_network_block: - step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=live_network_block, idempotency_key=idem)) - warnings.append(live_network_block) - continue - validation = await _validate_endpoint_target(test.url, request.external_host_allowlist) - if validation: - step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=validation, idempotency_key=idem)) - warnings.append(validation) - continue - result = await _call_endpoint_with_retries(test, request.max_retries, request.timeout_seconds, idem) - network_calls += result.attempts - step_results.append(result) - report = { - "integration_id": request.integration_id, - "dry_run": dry_run, - "allow_network": request.allow_network, - "network_calls_made": network_calls, - "generated_at": _now(), - "endpoint_results": [s.model_dump(mode="json") for s in step_results], - "contract_results": [c.model_dump(mode="json") for c in contract_results], - "safety": { - "ssrf_protection": "scheme, host allowlist, DNS/IP public-routability, and redirect revalidation enforced", - "secrets": "only secret references accepted; no raw values stored", - "payload_limits_bytes": MAX_JSON_BYTES, - "timeouts_seconds": request.timeout_seconds, - }, - } - path = f"{_integration_base(request.integration_id)}/acceptance-report.json" - artifact = await _write_workspace_json(ctx, path, report) - failed = any(s.status in {"fail", "blocked"} for s in step_results) or any(c.status == "fail" for c in contract_results) - audit.append(_audit("acceptance_completed", request.integration_id, {"report": path, "network_calls_made": network_calls, "failed_or_blocked": failed})) - await _emit(ctx, audit[-1]) - return RunAcceptanceResult( - status="blocked" if any(s.status == "blocked" for s in step_results) else "failed" if failed else "ok", - integration_id=request.integration_id, - dry_run=dry_run, - audit=audit, - warnings=sorted(set(warnings)), - residual_risks=["No production deployment, merge, or environment mutation is performed by this agent."], - acceptance_report_path=path, - artifacts=[artifact], - step_results=step_results, - contract_results=contract_results, - network_calls_made=network_calls, - ) + @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) - @a2a.tool(description="Prepare rollout, rollback, monitoring, and operator handoff documentation with approval-gated delivery planning", input_schema=_tool_input_schema(PrepareHandoffRequest), timeout_seconds=120, idempotent=True, max_retries=0, cost_class="deterministic") - async def prepare_handoff(self, ctx: RunContext[NoAuth], request: PrepareHandoffRequest) -> PrepareHandoffResult: - dry_run = _effective_dry_run(self, request.dry_run) - audit = [_audit("handoff_started", request.integration_id, {"dry_run": dry_run, "delivery_actions": len(request.delivery_actions)})] - await _emit(ctx, audit[-1]) - plan_material = { - "integration_id": request.integration_id, - "rollout_steps": request.rollout_steps, - "rollback_steps": request.rollback_steps, - "monitoring_checks": request.monitoring_checks, - "delivery_actions": request.delivery_actions, - } - digest = _plan_digest(plan_material) - approval_required = bool(request.delivery_actions) - approval_valid = (not approval_required) or ( - request.approve_delivery - and request.approval_phrase == APPROVAL_PHRASE - and request.plan_digest == digest - ) - warnings: list[str] = [] - if approval_required and not approval_valid: - warnings.append(f"delivery actions require approval_phrase={APPROVAL_PHRASE!r} and plan_digest={digest}") - if approval_required and approval_valid: - warnings.append("approval validated for the listed plan, but this agent still does not deploy, merge, or modify customer systems") - handoff_request = request.model_copy(update={"dry_run": dry_run}) - handoff = _render_handoff(handoff_request, digest, approval_required, approval_valid) - path = f"{_integration_base(request.integration_id)}/handoff.md" - artifact = await _write_workspace_text(ctx, path, handoff, "text/markdown") - audit.append(_audit("handoff_completed", request.integration_id, {"handoff": path, "approval_required": approval_required, "approval_valid": approval_valid})) - await _emit(ctx, audit[-1]) - return PrepareHandoffResult( - status="blocked" if approval_required and not approval_valid else "ok", - integration_id=request.integration_id, - dry_run=dry_run, - audit=audit, - warnings=warnings, - residual_risks=["Operators must execute rollout or rollback in caller-controlled systems outside this agent."], - handoff_path=path, - plan_digest=digest, - artifacts=[artifact], - approval_required=approval_required, - approval_valid=approval_valid, - delivery_executed=False, + 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, ) -def _now() -> str: - return datetime.now(UTC).isoformat() - - -def _effective_dry_run(agent: CustomerIntegrationEngineer, *values: bool) -> bool: - return bool(agent.config.default_dry_run or any(values)) - - -def _live_network_unavailable_reason(agent: CustomerIntegrationEngineer) -> str | None: - if getattr(agent.egress, "deny_internet_by_default", False): - return ( - "live network checks are disabled by this runtime because egress denies internet by default; " - "acceptance can validate metadata in dry_run mode but cannot contact external endpoints" - ) - return None - - -def _audit(action: str, integration_id: str, details: dict[str, Any] | None = None) -> AuditEntry: - return AuditEntry(ts=_now(), action=action, integration_id=integration_id, details=redact_obj(details or {})) - - -async def _emit(ctx: RunContext[NoAuth], entry: AuditEntry) -> None: - await ctx.emit_event(AgentEvent(kind="audit", payload=entry.model_dump(mode="json"))) - await ctx.emit_progress(f"{entry.action}: {entry.integration_id}") - - -def _integration_base(integration_id: str) -> str: - if not re.fullmatch(r"[a-z][a-z0-9-]{2,63}", integration_id): - raise ValueError("invalid integration_id") - return f"{OUTPUT_ROOT}/{integration_id}" - - -def _safe_output_path(path: str) -> str: - clean = path.replace("\\", "/").strip("/") - if not clean.startswith(f"{OUTPUT_ROOT}/"): - raise ValueError("writes must stay under outputs/integrations/{integration_id}/") - if ".." in clean.split("/") or clean.startswith("/"): - raise ValueError("unsafe output path") - return clean - - -async def _write_workspace_text(ctx: RunContext[NoAuth], path: str, text: str, mime_type: str) -> ArtifactRecord: - clean = _safe_output_path(path) - data = redact_secrets(text).encode("utf-8") - if len(data) > MAX_TEXT_BYTES * 4: - raise ValueError("artifact exceeds bounded text size") - ws = ctx.workspace - if hasattr(ws, "is_writable_output") and ws.is_writable_output(clean): # type: ignore[attr-defined] - if hasattr(ws, "write_bytes"): - ws.write_bytes(clean, data) # type: ignore[attr-defined] - else: - view = await ws.open_view(purpose=f"write {clean}", hints=[clean], max_files=1, mode=WorkspaceMode.READ_WRITE_OVERLAY, reason="Persist integration engineering output under outputs/integrations") - await view.write(clean, data) - else: - raise PermissionError("workspace grant does not allow writes under outputs/integrations") - artifact_name = clean.replace("/", "__") - ref = await ctx.write_artifact(artifact_name, data, mime_type) - await ctx.emit_artifact(ref) - return ArtifactRecord(path=clean, sha256=hashlib.sha256(data).hexdigest(), size_bytes=len(data), mime_type=mime_type) - - -async def _write_workspace_json(ctx: RunContext[NoAuth], path: str, payload: dict[str, Any]) -> ArtifactRecord: - safe_payload = redact_obj(payload) - _ensure_bounded_json(safe_payload, MAX_JSON_BYTES) - return await _write_workspace_text(ctx, path, json.dumps(safe_payload, indent=2, sort_keys=True) + "\n", "application/json") - - -def _ensure_bounded_json(value: Any, max_bytes: int) -> None: - data = json.dumps(value, default=str, sort_keys=True).encode("utf-8") - if len(data) > max_bytes: - raise ValueError(f"JSON payload exceeds {max_bytes} bytes") - - -def _reject_raw_secret(value: str) -> None: - if not value: - return - if any(marker in value.lower() for marker in ("sk-", "ghp_", "xoxb-", "-----begin", "bearer ")): - raise ValueError("raw credential-looking values are not allowed; use a secret_ref") - - -def redact_secrets(text: str) -> str: - redacted = text - patterns = [ - r"(?i)(authorization\s*[:=]\s*bearer\s+)[A-Za-z0-9._~+/=-]+", - r"(?i)(api[_-]?key\s*[:=]\s*)[A-Za-z0-9._~+/=-]{8,}", - r"(?i)(token\s*[:=]\s*)[A-Za-z0-9._~+/=-]{8,}", - r"(?i)(password\s*[:=]\s*)[^\s,;]+", - r"sk-[A-Za-z0-9]{12,}", - r"ghp_[A-Za-z0-9]{12,}", - ] - for pattern in patterns: - redacted = re.sub(pattern, _redact_match, redacted) - return redacted - - -def _redact_match(match: re.Match[str]) -> str: - return (match.group(1) if match.lastindex else "") + "[REDACTED]" - - -def _jsonschema_error_path(error: Any) -> Any: - return error.path - - -def redact_obj(value: Any) -> Any: - if isinstance(value, dict): - out: dict[str, Any] = {} - for key, item in value.items(): - if any(word in str(key).lower() for word in SECRET_WORDS): - out[key] = "[REDACTED_REF]" if isinstance(item, str) else "[REDACTED]" - else: - out[key] = redact_obj(item) - return out - if isinstance(value, list): - return [redact_obj(item) for item in value] - if isinstance(value, str): - return redact_secrets(value) - return value - - -def _plan_digest(payload: dict[str, Any]) -> str: - safe = redact_obj(payload) - encoded = json.dumps(safe, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") - return hashlib.sha256(encoded).hexdigest() - - -def _requirements_warnings(req: IntegrationRequirements) -> list[str]: - warnings: list[str] = [] - if not req.auth_methods: - warnings.append("no auth methods supplied; confirm this is intentional") - if not req.data_contracts: - warnings.append("no data contracts supplied; mapping and validation will be limited") - if not req.rate_limits: - warnings.append("no rate limits supplied; connector will use conservative defaults") - if any(env.purpose == "prod" and env.mutable for env in req.environments): - warnings.append("production environment marked mutable; this agent will still not mutate it") - return warnings - - -def _residual_risks(req: IntegrationRequirements) -> list[str]: - risks = [ - "Generated connector is minimal and must be reviewed by customer operators before production use.", - "Provider-specific authentication and observability wiring stay under caller-controlled setup references.", - ] - if req.dry_run: - risks.append("dry_run=true: acceptance evidence is deterministic and non-mutating, not proof of live provider behavior.") - return risks - - -def _required_setup(req: IntegrationRequirements) -> list[str]: - setup = {auth.secret_ref for auth in req.auth_methods if auth.secret_ref} - for env in req.environments: - for ref in (env.object_storage_ref, env.observability_ref): - if ref: - setup.add(ref) - return sorted(setup) - - -def _build_mapping(req: IntegrationRequirements) -> Any: - return { - "integration_id": req.integration_id, - "title": req.title, - "contracts": [ - { - "name": c.name, - "direction": c.direction, - "content_type": c.content_type, - "schema_ref": c.schema_ref, - "fields": [f.model_dump(mode="json") for f in c.fields], - } - for c in req.data_contracts - ], - "pii_fields": [f"{c.name}.{field.target}" for c in req.data_contracts for field in c.fields if field.pii], - } - - -def _render_architecture(req: IntegrationRequirements, digest: str, warnings: list[str], setup: list[str]) -> str: - systems = "\n".join(f"- **{s.name}** ({s.kind}, {s.environment}) {s.base_url or '(endpoint configured by caller)'}" for s in req.systems) - auth = "\n".join(f"- {a.system_name}: {a.auth_type}, secret_ref={a.secret_ref or 'none'}" for a in req.auth_methods) or "- No auth methods supplied." - contracts = "\n".join(f"- {c.name}: {c.direction}, {len(c.fields)} mapped fields, schema_ref={c.schema_ref or 'inline/fixture'}" for c in req.data_contracts) or "- No contracts supplied." - rates = "\n".join(f"- {r.system_name}: {r.max_requests}/{r.per_seconds}s burst={r.burst or r.max_requests}" for r in req.rate_limits) or "- Conservative default: bounded retries with idempotency keys." - compliance = "\n".join(f"- {c.category}: {c.name} — {c.requirement}" for c in req.compliance_constraints) or "- No explicit compliance constraints supplied." - success = "\n".join(f"- {s.name}: {s.measurement} target={s.target}" for s in req.success_criteria) - warning_text = "\n".join(f"- {w}" for w in warnings) or "- None" - setup_text = "\n".join(f"- {s}" for s in setup) or "- None" - return f"""# Integration Architecture: {req.title} - -Integration ID: `{req.integration_id}` -Plan digest: `{digest}` -Dry run: `{req.dry_run}` - -## Systems -{systems} - -## Authentication and Secret References -{auth} - -Raw credentials are neither accepted nor stored. Operators configure referenced secrets outside this agent. - -## Data Contracts and Field Mapping -{contracts} - -See `mapping.json` for exact field-level mapping. - -## Rate Limits and Retries -{rates} - -All mutating calls must carry idempotency keys. Retries are bounded and honor provider retry metadata. - -## Environments -{chr(10).join(f"- {e.name}: {e.purpose}, mutable={e.mutable}" for e in req.environments) or '- No environments supplied.'} - -## Compliance Constraints -{compliance} - -## Success Criteria -{success} - -## Required Caller Setup -{setup_text} - -## Safety Controls -- Default dry_run=true. -- External requests require explicit host allowlist, DNS/IP validation, redirect revalidation, payload limits, and timeouts. -- Repository writes beyond outputs, deployments, merges, and environment mutations are not performed. -- Delivery actions require `{APPROVAL_PHRASE}` bound to a plan digest and still remain operator-executed outside this agent. - -## Warnings -{warning_text} -""" - - -def _python_connector(req: IntegrationRequirements, allowlist: list[str], digest: str) -> str: - return f'''"""Minimal connector scaffold for {req.integration_id}. -Generated by customer-integration-engineer. Review before use. -Plan digest: {digest} -""" -from __future__ import annotations - -import hashlib -import ipaddress -import json -import socket -from urllib.parse import urlparse - -import httpx - -ALLOWED_HOSTS = {allowlist!r} -MAX_BYTES = {MAX_JSON_BYTES} -TIMEOUT_SECONDS = 8.0 - - -def validate_url(url: str) -> str: - parsed = urlparse(url) - if parsed.scheme != "https" or not parsed.hostname: - raise ValueError("connector only permits absolute https URLs") - host = parsed.hostname.lower() - if host not in ALLOWED_HOSTS: - raise ValueError(f"host {{host!r}} is not allowlisted") - for info in socket.getaddrinfo(host, parsed.port or 443, type=socket.SOCK_STREAM): - ip = ipaddress.ip_address(info[4][0]) - if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified: - raise ValueError(f"host {{host!r}} resolved to unsafe address") - return url - - -def idempotency_key(operation: str, payload: dict) -> str: - material = json.dumps({{"operation": operation, "payload": payload}}, sort_keys=True).encode("utf-8") - return hashlib.sha256(material).hexdigest() - - -async def send_json(url: str, payload: dict, *, secret_ref: str | None = None, dry_run: bool = True) -> dict: - body = json.dumps(payload).encode("utf-8") - if len(body) > MAX_BYTES: - raise ValueError("payload too large") - safe_url = validate_url(url) - key = idempotency_key("send_json", payload) - if dry_run: - return {{"status": "skipped", "reason": "dry_run", "idempotency_key": key, "url": safe_url}} - # Resolve secret_ref outside this scaffold using your approved secret manager. - headers = {{"Idempotency-Key": key, "Content-Type": "application/json"}} - async with httpx.AsyncClient(timeout=TIMEOUT_SECONDS, follow_redirects=False) as client: - response = await client.post(safe_url, content=body, headers=headers) - return {{"status_code": response.status_code, "idempotency_key": key}} -''' - - -def _python_connector_test(req: IntegrationRequirements) -> str: - return f'''from connector import idempotency_key - - -def test_idempotency_key_stable(): - payload = {{"integration_id": "{req.integration_id}", "event": "synthetic"}} - assert idempotency_key("send_json", payload) == idempotency_key("send_json", payload) -''' - - -def _typescript_connector(req: IntegrationRequirements, allowlist: list[str], digest: str) -> str: - hosts = json.dumps(allowlist) - return f'''// Minimal connector scaffold for {req.integration_id}. Plan digest: {digest} -import {{ lookup }} from "node:dns/promises"; -import net from "node:net"; - -const ALLOWED_HOSTS = new Set({hosts}); -const MAX_BYTES = {MAX_JSON_BYTES}; - -function isPublicAddress(address: string): boolean {{ - if (net.isIPv4(address)) {{ - const [a, b] = address.split(".").map(Number); - return !(a === 0 || a === 10 || a === 127 || a >= 224 || (a === 100 && b >= 64 && b <= 127) || (a === 169 && b === 254) || (a === 172 && b >= 16 && b <= 31) || (a === 192 && b === 168)); - }} - const lower = address.toLowerCase(); - return net.isIPv6(address) === 6 && !(lower === "::" || lower === "::1" || lower.startsWith("fe80:") || lower.startsWith("fc") || lower.startsWith("fd")); -}} - -export async function validateUrl(raw: string): Promise {{ - const url = new URL(raw); - if (url.protocol !== "https:") throw new Error("connector only permits https URLs"); - const host = url.hostname.toLowerCase(); - if (!ALLOWED_HOSTS.has(host)) throw new Error("host is not allowlisted"); - const records = await lookup(host, {{ all: true }}); - if (records.length === 0 || records.some(record => !isPublicAddress(record.address))) throw new Error("host resolved to unsafe address"); - return url; -}} - -export async function sendJson(rawUrl: string, payload: unknown, options: {{ dryRun?: boolean, secretRef?: string }} = {{}}) {{ - const url = await validateUrl(rawUrl); - const body = JSON.stringify(payload); - if (new TextEncoder().encode(body).length > MAX_BYTES) throw new Error("payload too large"); - const idem = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(body)); - const idempotencyKey = Array.from(new Uint8Array(idem)).map(b => b.toString(16).padStart(2, "0")).join(""); - if (options.dryRun !== false) return {{ status: "skipped", reason: "dry_run", idempotencyKey, url: url.toString() }}; - const response = await fetch(url, {{ method: "POST", redirect: "manual", body, headers: {{ "Content-Type": "application/json", "Idempotency-Key": idempotencyKey }} }}); - if (response.status >= 300 && response.status < 400) throw new Error("redirects are blocked"); - return {{ statusCode: response.status, idempotencyKey }}; -}} -''' - - -def _typescript_connector_test(req: IntegrationRequirements) -> str: - return f'''import {{ validateUrl }} from "./connector"; - -test("allowlist rejects unknown host for {req.integration_id}", () => {{ - expect(() => validateUrl("https://example.invalid/webhook")).toThrow(); -}}); -''' - - -def _validate_one_contract(contract: ContractArtifact) -> ContractCheck: - findings: list[str] = [] - fixture_results: list[dict[str, Any]] = [] - try: - if contract.kind == "json_schema": - from jsonschema import Draft202012Validator - Draft202012Validator.check_schema(contract.schema_doc) - validator = Draft202012Validator(contract.schema_doc) - for fixture in contract.fixtures: - errors = sorted(validator.iter_errors(fixture.payload), key=_jsonschema_error_path) - passed = not errors - expected = fixture.should_pass - fixture_results.append({"fixture": fixture.name, "passed": passed, "expected": expected, "errors": [e.message for e in errors[:5]]}) - if passed != expected: - findings.append(f"fixture {fixture.name!r} expected should_pass={expected} but got {passed}") - elif contract.kind == "openapi_3": - version = str(contract.schema_doc.get("openapi", "")) - if not version.startswith("3."): - findings.append("OpenAPI document must declare openapi: 3.x") - if not isinstance(contract.schema_doc.get("paths"), dict) or not contract.schema_doc.get("paths"): - findings.append("OpenAPI document must include non-empty paths") - if "servers" in contract.schema_doc and not isinstance(contract.schema_doc["servers"], list): - findings.append("OpenAPI servers must be a list") - elif contract.kind == "webhook": - if not contract.signature_header: - findings.append("webhook contract requires signature_header") - if not contract.secret_ref: - findings.append("webhook contract requires secret_ref, not a raw secret") - event_type = contract.schema_doc.get("event_type") or contract.schema_doc.get("type") - if not event_type: - findings.append("webhook schema_doc should include event_type or type") - for fixture in contract.fixtures: - has_event = "event" in fixture.payload or "type" in fixture.payload - passed = bool(has_event) - fixture_results.append({"fixture": fixture.name, "passed": passed, "expected": fixture.should_pass, "errors": [] if passed else ["missing event/type"]}) - if passed != fixture.should_pass: - findings.append(f"fixture {fixture.name!r} webhook expectation mismatch") - except Exception as exc: # noqa: BLE001 - findings.append(f"contract validation error: {type(exc).__name__}: {exc}") - status: Literal["pass", "fail", "warn"] = "fail" if findings else "pass" - return ContractCheck(name=contract.name, kind=contract.kind, status=status, findings=findings, fixture_results=fixture_results) - - -def _validate_hostname(host: str) -> str: - clean = host.strip().lower().rstrip(".") - if not clean or len(clean) > 253: - raise ValueError("invalid hostname") - if clean in {"localhost", "metadata.google.internal"} or clean.endswith(".local") or clean.endswith(".internal"): - raise ValueError("internal hostnames are not allowed") - try: - ip = ipaddress.ip_address(clean) - except ValueError: - if not re.fullmatch(r"[a-z0-9.-]+", clean) or ".." in clean: - raise ValueError("hostname contains invalid characters") - return clean - if _unsafe_ip(ip): - raise ValueError("private, loopback, link-local, multicast, reserved, or unspecified IPs are not allowed") - return clean - - -def _unsafe_ip(ip: ipaddress._BaseAddress) -> bool: - return bool(ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified) - - -def _validate_endpoint_metadata(url: str, allowlist: list[str]) -> str | None: - parsed = urlparse(url) - if parsed.scheme != "https" or not parsed.hostname: - return "endpoint URL must be absolute https with a hostname" - host = parsed.hostname.lower().rstrip(".") - if host not in set(allowlist): - return f"host {host!r} is not in external_host_allowlist" - try: - _validate_hostname(host) - except ValueError as exc: - return str(exc) - return None - - -async def _validate_endpoint_target(url: str, allowlist: list[str]) -> str | None: - metadata_error = _validate_endpoint_metadata(url, allowlist) - if metadata_error: - return metadata_error - parsed = urlparse(url) - host = parsed.hostname.lower().rstrip(".") # type: ignore[union-attr] - try: - infos = await asyncio.to_thread(socket.getaddrinfo, host, parsed.port or 443, type=socket.SOCK_STREAM) - except socket.gaierror as exc: - return f"DNS resolution failed for {host!r}: {exc}" - for info in infos: - ip = ipaddress.ip_address(info[4][0]) - if _unsafe_ip(ip): - return f"host {host!r} resolves to unsafe IP {ip}" - return None - - -async def _call_endpoint_with_retries(test: EndpointTest, max_retries: int, timeout_seconds: float, idem: str) -> AcceptanceStepResult: - import httpx - - attempts = 0 - last_detail = "not attempted" - headers = {"Idempotency-Key": idem, "Content-Type": "application/json"} - for attempt in range(max_retries + 1): - attempts = attempt + 1 - try: - async with httpx.AsyncClient(timeout=timeout_seconds, follow_redirects=False) as client: - response = await client.request(test.method, test.url, json=test.request_json, headers=headers) - if 300 <= response.status_code < 400: - location = response.headers.get("location", "") - return AcceptanceStepResult(name=test.name, status="blocked", attempts=attempts, detail=f"redirect blocked pending revalidation: {location[:200]}", idempotency_key=idem) - if response.status_code == test.expected_status: - return AcceptanceStepResult(name=test.name, status="pass", attempts=attempts, detail=f"received expected status {response.status_code}", idempotency_key=idem) - last_detail = f"expected {test.expected_status}, got {response.status_code}" - if response.status_code not in {408, 409, 425, 429, 500, 502, 503, 504}: - break - await asyncio.sleep(min(2.0, 0.25 * (attempt + 1))) - except Exception as exc: # noqa: BLE001 - last_detail = f"{type(exc).__name__}: {exc}" - await asyncio.sleep(min(2.0, 0.25 * (attempt + 1))) - return AcceptanceStepResult(name=test.name, status="fail", attempts=attempts, detail=redact_secrets(last_detail), idempotency_key=idem) - - -def _render_handoff(request: PrepareHandoffRequest, digest: str, approval_required: bool, approval_valid: bool) -> str: - operator_sections = "\n\n".join(f"## {s.heading}\n{s.body}" for s in request.operator_notes) or "## Operator Notes\nNo additional operator notes supplied." - delivery = "\n".join(f"- {item}" for item in request.delivery_actions) or "- No delivery actions requested." - return f"""# Integration Handoff: {request.title} - -Integration ID: `{request.integration_id}` -Plan digest: `{digest}` -Dry run: `{request.dry_run}` - -## Rollout Plan -{chr(10).join(f"{i+1}. {redact_secrets(step)}" for i, step in enumerate(request.rollout_steps))} - -## Rollback Plan -{chr(10).join(f"{i+1}. {redact_secrets(step)}" for i, step in enumerate(request.rollback_steps))} - -## Monitoring and Alerting Checks -{chr(10).join(f"- {redact_secrets(check)}" for check in request.monitoring_checks)} - -{operator_sections} - -## Delivery Actions Requiring Operator Control -{delivery} - -Approval required: `{approval_required}` -Approval valid for this exact digest: `{approval_valid}` - -This agent does not deploy, merge, modify customer systems, or execute delivery actions. Operators must perform approved delivery steps in their own controlled systems. - -## Evidence Checklist -- Review `architecture.md` and `mapping.json`. -- Review connector source under `connector/`. -- Review `contract-report.json`. -- Review `acceptance-report.json`. -- Confirm all secret references are configured in caller-owned systems. -""" +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 _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]: + """Copy packaged DeepAgents skills into the invocation workspace. + + DeepAgents loads skills from its backend, while source-controlled + ``skills/`` folders live in the image. This bridge lets generated agents + ship reusable SKILL.md bundles without giving up durable A2A workspace + files. + """ + root = Path(__file__).parent / "skills" + if not root.exists(): + return [] + runtime_skills_root = _runtime_skills_root(ctx) + uploads: list[tuple[str, bytes]] = [] + for path in root.rglob("*"): + if path.is_file(): + rel = path.relative_to(root).as_posix() + uploads.append((runtime_skills_root + rel, path.read_bytes())) + if uploads: + backend.upload_files(uploads) + return [runtime_skills_root] + return [] + + +def _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])