|
| 1 | +# At the top of the file, add: |
| 2 | +from typing import Annotated, Any, Literal, TypedDict |
| 3 | + |
| 4 | +from thirdweb_ai.services.service import Service |
| 5 | +from thirdweb_ai.tools.tool import tool |
| 6 | + |
| 7 | + |
| 8 | +class FilterField(TypedDict): |
| 9 | + field: Literal["id", "batchIndex", "from", "signerAddress", "smartAccountAddress", "chainId"] |
| 10 | + |
| 11 | + |
| 12 | +class FilterValues(TypedDict): |
| 13 | + values: list[int] |
| 14 | + |
| 15 | + |
| 16 | +class FilterOperator(TypedDict): |
| 17 | + operator: Literal["AND", "OR"] |
| 18 | + |
| 19 | + |
| 20 | +FilterCondition = FilterField | FilterValues | FilterOperator |
| 21 | + |
| 22 | + |
| 23 | +class EngineCloud(Service): |
| 24 | + def __init__( |
| 25 | + self, |
| 26 | + secret_key: str, |
| 27 | + vault_access_token: str, |
| 28 | + ): |
| 29 | + super().__init__(base_url="https://engine.thirdweb.com/v1", secret_key=secret_key) |
| 30 | + self.vault_access_token = vault_access_token |
| 31 | + |
| 32 | + def _make_headers(self): |
| 33 | + headers = super()._make_headers() |
| 34 | + if self.vault_access_token: |
| 35 | + headers["x-vault-access-token"] = self.vault_access_token |
| 36 | + return headers |
| 37 | + |
| 38 | + @tool( |
| 39 | + description="Create a new engine server wallet. This is a helper route for creating a new EOA with your KMS provider, provided as a convenient alternative to creating an EOA directly with your KMS provider." |
| 40 | + ) |
| 41 | + def create_server_wallet( |
| 42 | + self, |
| 43 | + label: Annotated[ |
| 44 | + str, |
| 45 | + "A human-readable label to identify this wallet.", |
| 46 | + ], |
| 47 | + ) -> dict[str, Any]: |
| 48 | + payload = {"label": label} |
| 49 | + return self._post("accounts", payload) |
| 50 | + |
| 51 | + @tool( |
| 52 | + description="Call a write function on a contract. This endpoint allows you to execute state-changing functions on smart contracts, with support for various execution strategies." |
| 53 | + ) |
| 54 | + def write_contract( |
| 55 | + self, |
| 56 | + from_address: Annotated[ |
| 57 | + str, |
| 58 | + "The address of the account to send the transaction from. Can be the address of a smart account or an EOA.", |
| 59 | + ], |
| 60 | + chain_id: Annotated[ |
| 61 | + int, |
| 62 | + "The numeric blockchain network ID where the contract is deployed (e.g., '1' for Ethereum mainnet, '137' for Polygon).", |
| 63 | + ], |
| 64 | + method: Annotated[str, "The name of the contract function to call on the contract."], |
| 65 | + params: Annotated[list[Any], "The arguments to pass to the contract function."], |
| 66 | + contract_address: Annotated[str, "The address of the smart contract to interact with."], |
| 67 | + abi: Annotated[list[dict[str, Any]], "The ABI (Application Binary Interface) of the contract."], |
| 68 | + value: Annotated[str, "The amount of native currency to send with the transaction (in wei)."] = "0", |
| 69 | + ) -> dict[str, Any]: |
| 70 | + payload = { |
| 71 | + "executionOptions": { |
| 72 | + "from": from_address, |
| 73 | + "chainId": chain_id, |
| 74 | + }, |
| 75 | + "params": [ |
| 76 | + { |
| 77 | + "method": method, |
| 78 | + "params": params, |
| 79 | + "contractAddress": contract_address, |
| 80 | + "abi": abi, |
| 81 | + "value": value, |
| 82 | + } |
| 83 | + ], |
| 84 | + } |
| 85 | + return self._post("write/contract", payload) |
| 86 | + |
| 87 | + @tool( |
| 88 | + description="Send an encoded transaction or a batch of transactions. This endpoint allows you to execute low-level transactions with raw transaction data." |
| 89 | + ) |
| 90 | + def send_transaction( |
| 91 | + self, |
| 92 | + from_address: Annotated[str, "The address of the account to send the transaction from."], |
| 93 | + chain_id: Annotated[ |
| 94 | + int, |
| 95 | + "The numeric blockchain network ID where the transaction will be sent (e.g., '1' for Ethereum mainnet, '137' for Polygon).", |
| 96 | + ], |
| 97 | + to_address: Annotated[ |
| 98 | + str, |
| 99 | + "The recipient address for the transaction.", |
| 100 | + ], |
| 101 | + data: Annotated[ |
| 102 | + str, |
| 103 | + "The encoded transaction data (hexadecimal).", |
| 104 | + ], |
| 105 | + value: Annotated[ |
| 106 | + str, |
| 107 | + "The amount of native currency to send with the transaction (in wei).", |
| 108 | + ] = "0", |
| 109 | + ) -> dict[str, Any]: |
| 110 | + payload = { |
| 111 | + "executionOptions": { |
| 112 | + "from": from_address, |
| 113 | + "chainId": chain_id, |
| 114 | + }, |
| 115 | + "params": [ |
| 116 | + { |
| 117 | + "to": to_address, |
| 118 | + "data": data, |
| 119 | + "value": value, |
| 120 | + } |
| 121 | + ], |
| 122 | + } |
| 123 | + |
| 124 | + return self._post("write/transaction", payload) |
| 125 | + |
| 126 | + @tool( |
| 127 | + description="Call read-only contract functions or batch read using multicall. This is a gas-efficient way to query data from blockchain contracts without modifying state." |
| 128 | + ) |
| 129 | + def read_contract( |
| 130 | + self, |
| 131 | + multicall_address: Annotated[ |
| 132 | + str | None, |
| 133 | + "Optional multicall contract address for batching multiple calls. Defaults to the default multicall3 address for the chain", |
| 134 | + ], |
| 135 | + chain_id: Annotated[ |
| 136 | + int, |
| 137 | + "The numeric blockchain network ID where the contract is deployed (e.g., '1' for Ethereum mainnet, '137' for Polygon).", |
| 138 | + ], |
| 139 | + from_address: Annotated[str, "EVM address in hex format"], |
| 140 | + method: Annotated[str, "The name of the contract function to call."], |
| 141 | + params: Annotated[list[Any], "The arguments to pass to the contract function."], |
| 142 | + contract_address: Annotated[ |
| 143 | + str, |
| 144 | + "The address of the smart contract to read from.", |
| 145 | + ], |
| 146 | + abi: Annotated[list[dict[str, Any]], "The ABI (Application Binary Interface) for the contract."], |
| 147 | + ) -> dict[str, Any]: |
| 148 | + payload = { |
| 149 | + "readOptions": { |
| 150 | + "multicallAddress": multicall_address, |
| 151 | + "chainId": chain_id, |
| 152 | + "from": from_address, |
| 153 | + }, |
| 154 | + "params": [ |
| 155 | + { |
| 156 | + "method": method, |
| 157 | + "params": params, |
| 158 | + "contractAddress": contract_address, |
| 159 | + "abi": abi, |
| 160 | + } |
| 161 | + ], |
| 162 | + } |
| 163 | + |
| 164 | + return self._post("read/contract", payload) |
| 165 | + |
| 166 | + @tool( |
| 167 | + description="Fetch the native cryptocurrency balance (e.g., ETH, MATIC) for a given address on a specific blockchain." |
| 168 | + ) |
| 169 | + def get_native_balance( |
| 170 | + self, |
| 171 | + chain_id: Annotated[ |
| 172 | + int, |
| 173 | + "The numeric blockchain network ID to query (e.g., '1' for Ethereum mainnet, '137' for Polygon).", |
| 174 | + ], |
| 175 | + address: Annotated[str, "The wallet address to check the balance for."], |
| 176 | + ) -> dict[str, Any]: |
| 177 | + payload = { |
| 178 | + "chainId": chain_id, |
| 179 | + "address": address, |
| 180 | + } |
| 181 | + |
| 182 | + return self._post("read/balance", payload) |
| 183 | + |
| 184 | + @tool( |
| 185 | + description="Search for transactions with flexible filtering options. Retrieve transaction history with customizable filters for addresses, chains, statuses, and more." |
| 186 | + ) |
| 187 | + def search_transactions( |
| 188 | + self, |
| 189 | + filters: Annotated[FilterField, "List of filter conditions to apply"], |
| 190 | + filters_operation: Annotated[ |
| 191 | + Literal["AND", "OR"], |
| 192 | + "Logical operation to apply between filters. 'AND' means all conditions must match, 'OR' means any condition can match.", |
| 193 | + ] = "AND", |
| 194 | + page: Annotated[ |
| 195 | + int | None, |
| 196 | + "Page number for paginated results, starting from 1.", |
| 197 | + ] = 1, |
| 198 | + limit: Annotated[ |
| 199 | + int | None, |
| 200 | + "Maximum number of transactions to return per page (1-100).", |
| 201 | + ] = 20, |
| 202 | + sort_by: Annotated[ |
| 203 | + Literal["createdAt", "confirmedAt"], |
| 204 | + "Field to sort results by.", |
| 205 | + ] = "createdAt", |
| 206 | + sort_direction: Annotated[ |
| 207 | + Literal["asc", "desc"], |
| 208 | + "Sort direction ('asc' for ascending, 'desc' for descending).", |
| 209 | + ] = "desc", |
| 210 | + ) -> dict[str, Any]: |
| 211 | + payload = { |
| 212 | + "filters": filters, |
| 213 | + "filtersOperation": filters_operation, |
| 214 | + "page": page, |
| 215 | + "limit": limit, |
| 216 | + "sortBy": sort_by, |
| 217 | + "sortDirection": sort_direction, |
| 218 | + } |
| 219 | + |
| 220 | + return self._post("transactions/search", payload) |
0 commit comments