feat(egress): add controlled_egress SDK helper emitting agent-signed egress_receipt/1

agentveil/__init__.py

@@ -13,6 +13,15 @@
 """
 
 from agentveil.agent import AVPAgent
+from agentveil.egress import (
+    ControlledEgressOutcome,
+    EgressPolicyViolationError,
+    EgressReceiptProofError,
+    EgressReceiptVerificationError,
+    controlled_egress,
+    sign_egress_receipt,
+    verify_egress_receipt,
+)
 from agentveil.mock import AVPMockAgent
 from agentveil.proof import ProofVerificationError, verify_proof_packet, verify_signed_jcs
 from agentveil.results import ControlledActionOutcome, IntegrationPreflightReport, ProofPacket
@@ -32,9 +41,16 @@
     "AVPAgent",
     "AVPMockAgent",
     "ControlledActionOutcome",
+    "ControlledEgressOutcome",
+    "EgressPolicyViolationError",
+    "EgressReceiptProofError",
+    "EgressReceiptVerificationError",
     "IntegrationPreflightReport",
     "ProofPacket",
     "ProofVerificationError",
+    "controlled_egress",
+    "sign_egress_receipt",
+    "verify_egress_receipt",
     "verify_proof_packet",
     "verify_signed_jcs",
     "avp_tracked",

agentveil/agent.py

@@ -19,7 +19,7 @@
 from copy import deepcopy
 from datetime import datetime, timedelta, timezone
 from importlib.metadata import PackageNotFoundError, version
-from typing import Any, Optional
+from typing import Any, Mapping, Optional
 
 import httpx
 from nacl.signing import SigningKey, VerifyKey
@@ -1623,6 +1623,48 @@ def controlled_action(
             reason=decision.get("reason", "runtime_gate_blocked"),
         )
 
+    def controlled_egress(
+        self,
+        *,
+        host: str,
+        port: int,
+        method: str,
+        path: str,
+        headers: Optional[Mapping[str, str]],
+        body: bytes,
+        credential_or_principal_class: str,
+        policy_id: str,
+        delegation_receipt: Mapping[str, Any],
+        timeout_seconds: float = 30.0,
+    ) -> "ControlledEgressOutcome":
+        """Perform an HTTPS egress through the AVP-controlled boundary.
+
+        Returns a ``ControlledEgressOutcome`` carrying an **agent-signed**
+        ``egress_receipt/1`` for ALLOW/BLOCK/FAILED outcomes (no receipt
+        on ``WAITING_FOR_HUMAN_APPROVAL`` in v0.1). The helper performs
+        the HTTPS send itself; calls that bypass this method are not
+        recorded.
+
+        Receipts are signed by the agent's own identity, not by the AVP
+        backend. Verify with
+        ``verify_egress_receipt(receipt_jcs, trusted_signer_dids=[self.did])``.
+        """
+        from agentveil.egress import controlled_egress as _controlled_egress
+
+        return _controlled_egress(
+            agent=self,
+            host=host,
+            port=port,
+            method=method,
+            path=path,
+            headers=headers,
+            body=body,
+            credential_or_principal_class=credential_or_principal_class,
+            policy_id=policy_id,
+            delegation_receipt=delegation_receipt,
+            timeout_seconds=timeout_seconds,
+        )
+
     def execute_after_approval(
         self,
         audit_id: str,