add engine cloud service

README.md

@@ -4,7 +4,7 @@ _AI Agents with Onchain Intelligence_
 
 ## 📖 Overview
 
-thirdweb AI is thirdweb's comprehensive toolkit for blockchain data analysis, wallet management, and AI agent interaction with blockchains. It simplifies complex blockchain operations into four core components: Insight for data analysis, Engine for wallet and contract operations, Storage for decentralized file management, and Nebula for natural language-powered blockchain interactions.
+thirdweb AI is thirdweb's comprehensive toolkit for blockchain data analysis, wallet management, and AI agent interaction with blockchains. It simplifies complex blockchain operations into five core components: Insight for data analysis, Engine for wallet and contract operations, EngineCloud for cloud-based engine operations, Storage for decentralized file management, and Nebula for natural language-powered blockchain interactions.
 
 ## 🌐 Features
 
@@ -23,6 +23,13 @@ Core blockchain interaction capabilities:
 - **Read**: Read operations for smart contracts and blockchain data
 - **Write**: Transaction creation and contract interaction
 
+### EngineCloud
+Cloud-based engine operations with advanced capabilities:
+- **Server Wallets**: Create and manage server wallets with KMS integration
+- **Contract Interaction**: Read from and write to smart contracts
+- **Transaction Management**: Send transactions and query transaction history
+- **Balance Queries**: Check native token balances on various chains
+
 ### Storage
 Decentralized storage capabilities:
 - **Upload**: Upload files, directories, and JSON data to IPFS
@@ -73,12 +80,13 @@ See the list of [supported framework and installation guides](python/thirdweb-ai
 #### Basic Usage
 
 ```python
-from thirdweb_ai import Engine, Insight, Nebula, Storage, Tool
+from thirdweb_ai import Engine, EngineCloud, Insight, Nebula, Storage, Tool
 
 # Initialize services
 insight = Insight(secret_key=...)
 nebula = Nebula(secret_key=...)
 engine = Engine(...)
+engine_cloud = EngineCloud(secret_key=..., vault_access_token=...)  # For cloud-based operations
 storage = Storage(secret_key=...)
 
 # Example: Create tools for AI agents
@@ -92,6 +100,7 @@ tools = [
 # tools = [
 #     *insight.get_tools(),
 #     *engine.get_tools(),
+#     *engine_cloud.get_tools(),
 #     *storage.get_tools(),
 # ]
 
@@ -128,7 +137,7 @@ For non-security-related bugs, please use the GitHub issue tracker.
 
 ## ⚠️ Important Usage Notes
 
-When using Nebula, do not combine it with other tools (Insight, Engine, Storage) in the same agent implementation as Nebula already calls these tools in the background. Using them together can lead to compatibility issues and unexpected behavior.
+When using Nebula, do not combine it with other tools (Insight, Engine, EngineCloud, Storage) in the same agent implementation as Nebula already calls these tools in the background. Using them together can lead to compatibility issues and unexpected behavior.
 
 ## 📦 Publishing Workflow
 
@@ -149,4 +158,4 @@ To publish a new version of thirdweb AI packages:
 
 ## 📝 License
 
-thirdweb AI is licensed under the Apache-2.0 License. See the [LICENSE](./LICENSE) file for details.
+thirdweb AI is licensed under the Apache-2.0 License. See the [LICENSE](./LICENSE) file for details.

python/thirdweb-ai/README.md

@@ -49,12 +49,13 @@ pip install "thirdweb-ai[pydantic-ai]"  # For Pydantic AI
 thirdweb-ai provides a set of tools that can be integrated with various AI agent frameworks. Here's a basic example:
 
 ```python
-from thirdweb_ai import Engine, Insight, Nebula, Storage, Tool
+from thirdweb_ai import Engine, EngineCloud, Insight, Nebula, Storage, Tool
 
 # Initialize thirdweb services
 insight = Insight(secret_key=...)
 nebula = Nebula(secret_key=...)
 engine = Engine(secret_key=...)
+engine_cloud = EngineCloud(secret_key=..., vault_access_token=...)  # vault_access_token required for server wallet operations
 storage = Storage(secret_key=...)
 
 # Get available tools
@@ -64,11 +65,22 @@ tools = [
     *insight.get_tools(),
     *nebula.get_tools(),
     *engine.get_tools(),
+    *engine_cloud.get_tools(),  # Use EngineCloud for cloud-based engine operations
     *storage.get_tools(),
     # Or pick an individual tool from the services
 ]
 ```
 
+### Available Services
+
+thirdweb-ai provides several core services:
+
+- **Engine**: Deploy contracts, manage wallets, execute transactions, and interact with smart contracts
+- **EngineCloud**: Cloud-based engine operations for creating server wallets (with KMS integration), executing contract calls, and querying transaction history
+- **Insight**: Query blockchain data, retrieve transactions, events, token balances, and contract metadata
+- **Nebula**: Advanced onchain analytics and data processing
+- **Storage**: Store and retrieve data using IPFS and other decentralized storage solutions
+
 ## Framework Integration
 
 thirdweb-ai can be easily integrated with various AI agent frameworks. Below are a few examples of how to integrate with some of the supported frameworks:
@@ -191,4 +203,4 @@ uv run ruff check .
 
 # Run type checking with pyright
 uv run pyright
-```
+```

python/thirdweb-ai/src/thirdweb_ai/adapters/google_adk/__init__.py

@@ -0,0 +1,3 @@
+from .google_adk import get_google_adk_tools
+
+__all__ = ["get_google_adk_tools"]

python/thirdweb-ai/src/thirdweb_ai/adapters/google_adk/google_adk.py

@@ -40,8 +40,29 @@ def _get_declaration(self) -> types.FunctionDeclaration:
         Returns:
             A FunctionDeclaration for Google ADK
         """
-        parameters = self.tool.schema["parameters"]
-        del parameters["additionalProperties"]
+        # Deep copy the parameters to avoid modifying the original
+        import copy
+        parameters = copy.deepcopy(self.tool.schema["parameters"])
+
+        if "additionalProperties" in parameters:
+            del parameters["additionalProperties"]
+
+        def remove_additional_properties(obj: dict[str, Any]):
+            if "additionalProperties" in obj:
+                del obj["additionalProperties"]
+
+            if "items" in obj and isinstance(obj["items"], dict):
+                remove_additional_properties(obj["items"])
+
+            if "properties" in obj and isinstance(obj["properties"], dict):
+                for prop in obj["properties"].values():
+                    if isinstance(prop, dict):
+                        remove_additional_properties(prop)
+
+        if "properties" in parameters:
+            for prop in parameters["properties"].values():
+                remove_additional_properties(prop)
+
         return types.FunctionDeclaration(
             name=self.name,
             description=self.description,

python/thirdweb-ai/src/thirdweb_ai/services/__init__.py

@@ -0,0 +1,8 @@
+from thirdweb_ai.services.engine import Engine
+from thirdweb_ai.services.engine_cloud import EngineCloud
+from thirdweb_ai.services.insight import Insight
+from thirdweb_ai.services.nebula import Nebula
+from thirdweb_ai.services.service import Service
+from thirdweb_ai.services.storage import Storage
+
+__all__ = ["Engine", "EngineCloud", "Insight", "Nebula", "Service", "Storage"]

python/thirdweb-ai/src/thirdweb_ai/services/engine_cloud.py

@@ -0,0 +1,220 @@
+# At the top of the file, add:
+from typing import Annotated, Any, Literal, TypedDict
+
+from thirdweb_ai.services.service import Service
+from thirdweb_ai.tools.tool import tool
+
+
+class FilterField(TypedDict):
+    field: Literal["id", "batchIndex", "from", "signerAddress", "smartAccountAddress", "chainId"]
+
+
+class FilterValues(TypedDict):
+    values: list[int]
+
+
+class FilterOperator(TypedDict):
+    operator: Literal["AND", "OR"]
+
+
+FilterCondition = FilterField | FilterValues | FilterOperator
+
+
+class EngineCloud(Service):
+    def __init__(
+        self,
+        secret_key: str,
+        vault_access_token: str,
+    ):
+        super().__init__(base_url="https://engine.thirdweb.com/v1", secret_key=secret_key)
+        self.vault_access_token = vault_access_token
+
+    def _make_headers(self):
+        headers = super()._make_headers()
+        if self.vault_access_token:
+            headers["x-vault-access-token"] = self.vault_access_token
+        return headers
+
+    @tool(
+        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."
+    )
+    def create_server_wallet(
+        self,
+        label: Annotated[
+            str,
+            "A human-readable label to identify this wallet.",
+        ],
+    ) -> dict[str, Any]:
+        payload = {"label": label}
+        return self._post("accounts", payload)
+
+    @tool(
+        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."
+    )
+    def write_contract(
+        self,
+        from_address: Annotated[
+            str,
+            "The address of the account to send the transaction from. Can be the address of a smart account or an EOA.",
+        ],
+        chain_id: Annotated[
+            int,
+            "The numeric blockchain network ID where the contract is deployed (e.g., '1' for Ethereum mainnet, '137' for Polygon).",
+        ],
+        method: Annotated[str, "The name of the contract function to call on the contract."],
+        params: Annotated[list[Any], "The arguments to pass to the contract function."],
+        contract_address: Annotated[str, "The address of the smart contract to interact with."],
+        abi: Annotated[list[dict[str, Any]], "The ABI (Application Binary Interface) of the contract."],
+        value: Annotated[str, "The amount of native currency to send with the transaction (in wei)."] = "0",
+    ) -> dict[str, Any]:
+        payload = {
+            "executionOptions": {
+                "from": from_address,
+                "chainId": chain_id,
+            },
+            "params": [
+                {
+                    "method": method,
+                    "params": params,
+                    "contractAddress": contract_address,
+                    "abi": abi,
+                    "value": value,
+                }
+            ],
+        }
+        return self._post("write/contract", payload)
+
+    @tool(
+        description="Send an encoded transaction or a batch of transactions. This endpoint allows you to execute low-level transactions with raw transaction data."
+    )
+    def send_transaction(
+        self,
+        from_address: Annotated[str, "The address of the account to send the transaction from."],
+        chain_id: Annotated[
+            int,
+            "The numeric blockchain network ID where the transaction will be sent (e.g., '1' for Ethereum mainnet, '137' for Polygon).",
+        ],
+        to_address: Annotated[
+            str,
+            "The recipient address for the transaction.",
+        ],
+        data: Annotated[
+            str,
+            "The encoded transaction data (hexadecimal).",
+        ],
+        value: Annotated[
+            str,
+            "The amount of native currency to send with the transaction (in wei).",
+        ] = "0",
+    ) -> dict[str, Any]:
+        payload = {
+            "executionOptions": {
+                "from": from_address,
+                "chainId": chain_id,
+            },
+            "params": [
+                {
+                    "to": to_address,
+                    "data": data,
+                    "value": value,
+                }
+            ],
+        }
+
+        return self._post("write/transaction", payload)
+
+    @tool(
+        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."
+    )
+    def read_contract(
+        self,
+        multicall_address: Annotated[
+            str | None,
+            "Optional multicall contract address for batching multiple calls. Defaults to the default multicall3 address for the chain",
+        ],
+        chain_id: Annotated[
+            int,
+            "The numeric blockchain network ID where the contract is deployed (e.g., '1' for Ethereum mainnet, '137' for Polygon).",
+        ],
+        from_address: Annotated[str, "EVM address in hex format"],
+        method: Annotated[str, "The name of the contract function to call."],
+        params: Annotated[list[Any], "The arguments to pass to the contract function."],
+        contract_address: Annotated[
+            str,
+            "The address of the smart contract to read from.",
+        ],
+        abi: Annotated[list[dict[str, Any]], "The ABI (Application Binary Interface) for the contract."],
+    ) -> dict[str, Any]:
+        payload = {
+            "readOptions": {
+                "multicallAddress": multicall_address,
+                "chainId": chain_id,
+                "from": from_address,
+            },
+            "params": [
+                {
+                    "method": method,
+                    "params": params,
+                    "contractAddress": contract_address,
+                    "abi": abi,
+                }
+            ],
+        }
+
+        return self._post("read/contract", payload)
+
+    @tool(
+        description="Fetch the native cryptocurrency balance (e.g., ETH, MATIC) for a given address on a specific blockchain."
+    )
+    def get_native_balance(
+        self,
+        chain_id: Annotated[
+            int,
+            "The numeric blockchain network ID to query (e.g., '1' for Ethereum mainnet, '137' for Polygon).",
+        ],
+        address: Annotated[str, "The wallet address to check the balance for."],
+    ) -> dict[str, Any]:
+        payload = {
+            "chainId": chain_id,
+            "address": address,
+        }
+
+        return self._post("read/balance", payload)
+
+    @tool(
+        description="Search for transactions with flexible filtering options. Retrieve transaction history with customizable filters for addresses, chains, statuses, and more."
+    )
+    def search_transactions(
+        self,
+        filters: Annotated[FilterField, "List of filter conditions to apply"],
+        filters_operation: Annotated[
+            Literal["AND", "OR"],
+            "Logical operation to apply between filters. 'AND' means all conditions must match, 'OR' means any condition can match.",
+        ] = "AND",
+        page: Annotated[
+            int | None,
+            "Page number for paginated results, starting from 1.",
+        ] = 1,
+        limit: Annotated[
+            int | None,
+            "Maximum number of transactions to return per page (1-100).",
+        ] = 20,
+        sort_by: Annotated[
+            Literal["createdAt", "confirmedAt"],
+            "Field to sort results by.",
+        ] = "createdAt",
+        sort_direction: Annotated[
+            Literal["asc", "desc"],
+            "Sort direction ('asc' for ascending, 'desc' for descending).",
+        ] = "desc",
+    ) -> dict[str, Any]:
+        payload = {
+            "filters": filters,
+            "filtersOperation": filters_operation,
+            "page": page,
+            "limit": limit,
+            "sortBy": sort_by,
+            "sortDirection": sort_direction,
+        }
+
+        return self._post("transactions/search", payload)