Adapt SDK to token-based deploy API, add client-side verification

API changes:
- Prepare endpoint: GET with URL params → POST with JSON body
- Submit payload: token field replaces prepare_response
- Response schema: token replaces server_signature

Security verification before signing:
- Layer 1: factory set (is deployment), sender matches agent_address, key in owners
- Layer 2: compute EIP-4337 v0.7 UserOp hash locally, compare with server hash

Prevents signing malicious operations even if server is compromised.

Author: matiasedgeandnode

python/ampersend-sdk/src/ampersend_sdk/ampersend/management.py

@@ -4,13 +4,19 @@
 from typing import Any, Dict, List, Optional, Self
 
 import httpx
+from eth_abi.abi import encode
 from eth_account import Account
 from eth_account.messages import encode_defunct
+from eth_utils.crypto import keccak
 from pydantic import BaseModel, ConfigDict, Field
 
 from .types import ApiError
 
 DEFAULT_API_URL = "https://api.ampersend.ai"
+DEFAULT_CHAIN_ID = 84532  # Base Sepolia
+
+# ERC-4337 EntryPoint v0.7 address (same on all chains)
+ENTRYPOINT_V07_ADDRESS = "0x0000000071727De22E5E9d8BAf0edAc6f37da032"
 
 
 class SpendConfig(BaseModel):
@@ -84,10 +90,12 @@ def __init__(
         api_key: str,
         api_url: str = DEFAULT_API_URL,
         timeout: float = 30,
+        chain_id: int = DEFAULT_CHAIN_ID,
     ) -> None:
         self._api_key = api_key
         self._base_url = api_url.rstrip("/")
         self._timeout = timeout
+        self._chain_id = chain_id
         self._http_client: Optional[httpx.AsyncClient] = None
 
     async def __aenter__(self) -> Self:
@@ -122,6 +130,8 @@ async def create_agent(
         only locally to derive the address and sign the deployment UserOp; it is
         never sent to the server.
 
+        Verifies the server response before signing to prevent malicious operations.
+
         Args:
             name: Human-readable agent name.
             private_key: Agent owner private key (hex, 0x-prefixed).
@@ -134,33 +144,176 @@ async def create_agent(
         account = Account.from_key(private_key)
         agent_key_address = account.address
 
-        # 1. Prepare unsigned UserOp
+        # 1. Prepare unsigned UserOp (POST with JSON body)
         prepare_response = await self._fetch(
-            "GET",
+            "POST",
             "/api/v1/sdk/agents/prepare",
-            params={"agent_key_address": agent_key_address},
+            json_data={"agent_key_address": agent_key_address},
         )
 
-        # 2. Sign the UserOp hash
+        # 2. Verify the response before signing (Layer 1: basic sanity checks)
+        self._verify_prepare_response(prepare_response, agent_key_address)
+
+        # 3. Sign the UserOp hash
         user_op_hash: str = prepare_response["user_op_hash"]
         hash_bytes = bytes.fromhex(user_op_hash.removeprefix("0x"))
         signed = account.sign_message(encode_defunct(primitive=hash_bytes))
         signature = "0x" + signed.signature.hex()
 
-        # 3. Build create payload
+        # 4. Build create payload (token-based, not full prepare_response)
         payload = {
+            "token": prepare_response["token"],
             "signature": signature,
-            "prepare_response": prepare_response,
             "name": name,
             "keys": [{"address": agent_key_address, "permission_id": None}],
             "spend_config": spend_config.to_api_dict() if spend_config else None,
             "authorized_sellers": authorized_sellers,
         }
 
-        # 4. Submit signed deployment
+        # 5. Submit signed deployment
         response = await self._fetch("POST", "/api/v1/sdk/agents", json_data=payload)
         return AgentResponse(**response)
 
+    def _verify_prepare_response(
+        self, response: Dict[str, Any], agent_key_address: str
+    ) -> None:
+        """Verify the prepare response before signing.
+
+        Layer 1: Basic sanity checks to prevent signing malicious operations.
+        Layer 2: Hash verification to ensure we sign what we expect.
+        """
+        user_op = response.get("unsigned_user_op", {})
+
+        # Layer 1, Check 1: This is a deployment (factory must be set)
+        if user_op.get("factory") is None:
+            raise ApiError(
+                "Invalid prepare response: not a deployment operation (factory is null)"
+            )
+
+        # Layer 1, Check 2: Deploying the expected address
+        sender = user_op.get("sender")
+        agent_address = response.get("agent_address", "")
+        if not sender or sender.lower() != agent_address.lower():
+            raise ApiError(
+                f"Invalid prepare response: sender mismatch "
+                f"(expected {agent_address}, got {sender})"
+            )
+
+        # Layer 1, Check 3: Our key is in the owners list
+        owners = response.get("owners", [])
+        owner_addresses = [o.lower() for o in owners]
+        if agent_key_address.lower() not in owner_addresses:
+            raise ApiError(
+                f"Invalid prepare response: agent key {agent_key_address} "
+                f"not in owners list"
+            )
+
+        # Layer 2: Verify hash matches the UserOp we're signing
+        computed_hash = self._compute_user_op_hash(user_op)
+        server_hash = response.get("user_op_hash", "")
+        if computed_hash.lower() != server_hash.lower():
+            raise ApiError(
+                f"Invalid prepare response: hash mismatch "
+                f"(computed {computed_hash}, server provided {server_hash})"
+            )
+
+    def _compute_user_op_hash(self, user_op: Dict[str, Any]) -> str:
+        """Compute ERC-4337 v0.7 UserOperation hash.
+
+        Reference: https://eips.ethereum.org/EIPS/eip-4337
+        """
+
+        def to_bytes(hex_str: Optional[str]) -> bytes:
+            if not hex_str:
+                return b""
+            return bytes.fromhex(hex_str.removeprefix("0x"))
+
+        def to_int(value: Any) -> int:
+            if isinstance(value, int):
+                return value
+            if isinstance(value, str):
+                return int(value, 16) if value.startswith("0x") else int(value)
+            return 0
+
+        def to_address(addr: Optional[str]) -> bytes:
+            if not addr:
+                return b"\x00" * 20
+            return bytes.fromhex(addr.removeprefix("0x").zfill(40))
+
+        # Extract fields
+        sender = to_address(user_op.get("sender"))
+        nonce = to_int(user_op.get("nonce", 0))
+        factory = user_op.get("factory")
+        factory_data = user_op.get("factoryData", "0x")
+        call_data = to_bytes(user_op.get("callData", "0x"))
+        call_gas_limit = to_int(user_op.get("callGasLimit", 0))
+        verification_gas_limit = to_int(user_op.get("verificationGasLimit", 0))
+        pre_verification_gas = to_int(user_op.get("preVerificationGas", 0))
+        max_fee_per_gas = to_int(user_op.get("maxFeePerGas", 0))
+        max_priority_fee_per_gas = to_int(user_op.get("maxPriorityFeePerGas", 0))
+        paymaster = user_op.get("paymaster")
+        paymaster_verification_gas_limit = to_int(
+            user_op.get("paymasterVerificationGasLimit", 0)
+        )
+        paymaster_post_op_gas_limit = to_int(user_op.get("paymasterPostOpGasLimit", 0))
+        paymaster_data = to_bytes(user_op.get("paymasterData", "0x"))
+
+        # Build initCode: factory + factoryData (or empty)
+        if factory:
+            init_code = to_address(factory) + to_bytes(factory_data)
+        else:
+            init_code = b""
+
+        # Build paymasterAndData: paymaster + gasLimits (16 bytes each) + data
+        if paymaster:
+            paymaster_and_data = (
+                to_address(paymaster)
+                + paymaster_verification_gas_limit.to_bytes(16, "big")
+                + paymaster_post_op_gas_limit.to_bytes(16, "big")
+                + paymaster_data
+            )
+        else:
+            paymaster_and_data = b""
+
+        # Pack gas limits: (verificationGasLimit << 128) | callGasLimit
+        account_gas_limits = (verification_gas_limit << 128) | call_gas_limit
+
+        # Pack gas fees: (maxPriorityFeePerGas << 128) | maxFeePerGas
+        gas_fees = (max_priority_fee_per_gas << 128) | max_fee_per_gas
+
+        # Encode inner struct (v0.7 format)
+        inner_encoded = encode(
+            [
+                "address",
+                "uint256",
+                "bytes32",
+                "bytes32",
+                "bytes32",
+                "uint256",
+                "bytes32",
+                "bytes32",
+            ],
+            [
+                sender,
+                nonce,
+                keccak(init_code),
+                keccak(call_data),
+                account_gas_limits.to_bytes(32, "big"),
+                pre_verification_gas,
+                gas_fees.to_bytes(32, "big"),
+                keccak(paymaster_and_data),
+            ],
+        )
+
+        # Final hash: keccak(keccak(innerEncoded), entryPoint, chainId)
+        entrypoint = to_address(ENTRYPOINT_V07_ADDRESS)
+        final_encoded = encode(
+            ["bytes32", "address", "uint256"],
+            [keccak(inner_encoded), entrypoint, self._chain_id],
+        )
+
+        return "0x" + keccak(final_encoded).hex()
+
     async def list_agents(self) -> List[AgentResponse]:
         """List all agents belonging to the authenticated user."""
         result = await self._fetch("GET", "/api/v1/sdk/agents")