lint with ruff and fix types

Author: cjber

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

@@ -1,11 +1,11 @@
+import copy
 from typing import Any
 
 from google.adk.tools import BaseTool, ToolContext
 from google.genai import types
 
 from thirdweb_ai.tools.tool import Tool, ToolSchema
 
-from pydantic import BaseModel
 
 class GoogleAdkTool(BaseTool):
     """Adapter for Thirdweb Tool to Google ADK BaseTool.
@@ -40,38 +40,34 @@ def _get_declaration(self) -> types.FunctionDeclaration:
         Returns:
             A FunctionDeclaration for Google ADK
         """
-        # 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)
-        
+
+        if "parameters" not in self.tool.schema:
+            raise ValueError("Tool schema must contain 'parameters'.")
+
+        # Create a clean parameters dict without additionalProperties
+        parameters = copy.deepcopy(dict(self.tool.schema["parameters"]))
+
+        # Remove additionalProperties recursively from the entire schema
+        def clean_schema(obj: dict[str, Any]) -> dict[str, Any]:
+            cleaned = {k: v for k, v in obj.items() if k != "additionalProperties"}
+
+            for key, value in cleaned.items():
+                if isinstance(value, dict):
+                    cleaned[key] = clean_schema(value)
+                elif isinstance(value, list):
+                    cleaned[key] = [clean_schema(item) if isinstance(item, dict) else item for item in value]
+
+            return cleaned
+
+        clean_parameters = clean_schema(parameters)
+
         return types.FunctionDeclaration(
             name=self.name,
             description=self.description,
-            parameters=parameters,
+            parameters=types.Schema(**clean_parameters),
         )
 
-    # Override the method with the expected signature based on the error message
-    # and adapting from the reference implementation
-    async def run_async(self, args: dict[str, Any], tool_context: ToolContext) -> Any:
+    async def run_async(self, *, args: dict[str, Any], tool_context: ToolContext) -> Any:
         """Execute the tool asynchronously.
 
         This method adapts the Thirdweb tool to work with Google ADK's async execution.

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

@@ -1,3 +1,5 @@
+from typing import Any, cast
+
 import mcp.types as types
 from mcp.server.fastmcp import FastMCP
 from mcp.server.fastmcp.tools.base import Tool as FastMCPTool
@@ -11,26 +13,27 @@ def get_fastmcp_tools(tools: list[Tool]) -> list[FastMCPTool]:
             fn=lambda _t=tool, **kwargs: _t.run_json(kwargs),
             name=tool.name,
             description=tool.description,
-            parameters=tool.schema.get("parameters"),
+            parameters=cast(dict[str, Any], tool.schema.get("parameters") or {}),
             fn_metadata=func_metadata(tool._func_definition, skip_names=["self"]),  # noqa: SLF001
             is_async=False,
             context_kwarg=None,
+            annotations=tool.annotations,
         )
         for tool in tools
     ]
 
 
 def add_fastmcp_tools(fastmcp: FastMCP, tools: list[Tool]):
     for tool in get_fastmcp_tools(tools):
-        fastmcp._tool_manager._tools[tool.name] = tool  # noqa: SLF001
+        fastmcp._tool_manager._tools[tool.name] = tool  # type: ignore[reportPrivateUsage]  # noqa: SLF001
 
 
 def get_mcp_tools(tools: list[Tool]) -> list[types.Tool]:
     return [
         types.Tool(
             name=tool.name,
             description=tool.description,
-            inputSchema=tool.schema.get("parameters"),
+            inputSchema=cast(dict[str, Any], tool.schema.get("parameters") or {}),
         )
         for tool in tools
     ]

python/thirdweb-ai/src/thirdweb_ai/adapters/openai/agents.py

@@ -29,15 +29,15 @@ def _get_openai_schema(schema: Any):
 
 def get_agents_tools(tools: list[Tool]):
     def _get_tool(tool: Tool):
-        async def _invoke_tool(ctx: RunContextWrapper[Any], tool_input: str, _t: Tool = tool) -> str:  # type: ignore # noqa: PGH003
+        async def _invoke_tool(ctx: RunContextWrapper[Any], tool_input: str, _t: Tool = tool) -> str:
             input_args = json.loads(tool_input) if tool_input else {}
             return _t.return_value_as_string(_t.run_json(input_args))
 
         schema = _get_openai_schema(tool.schema.get("parameters"))
-        return FunctionTool(  # type: ignore # noqa: PGH003
+        return FunctionTool(
             name=tool.name,
             description=tool.description,
-            params_json_schema=schema,
+            params_json_schema=schema or {},
             on_invoke_tool=_invoke_tool,
         )
 

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

@@ -12,6 +12,8 @@ async def execute(**kwargs: Any) -> Any:
             return tool.run_json(kwargs)
 
         async def prepare(ctx: RunContext, tool_def: ToolDefinition) -> ToolDefinition:
+            if "parameters" not in tool.schema:
+                raise ValueError("Tool schema must contain 'parameters'.")
             tool_def.parameters_json_schema = tool.schema["parameters"]
             return tool_def
 

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

@@ -5,4 +5,4 @@
 from thirdweb_ai.services.service import Service
 from thirdweb_ai.services.storage import Storage
 
-__all__ = ["Engine", "EngineCloud", "Insight", "Nebula", "Service", "Storage"]
+__all__ = ["Engine", "EngineCloud", "Insight", "Nebula", "Service", "Storage"]

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

@@ -218,3 +218,14 @@ def search_transactions(
         }
 
         return self._post("transactions/search", payload)
+
+    @tool(
+        description="List all engine server wallets for the current project. Returns an array of EOA addresses with their corresponding predicted smart account addresses."
+    )
+    def get_accounts(self) -> dict[str, Any]:
+        """Get all engine server wallets for the current project.
+
+        Returns:
+            dict containing list of account objects with address and smartAccountAddress
+        """
+        return self._get("accounts")

python/thirdweb-ai/src/thirdweb_ai/tools/tool.py

@@ -72,7 +72,7 @@ def args_base_model_from_signature(name: str, sig: inspect.Signature) -> type[Ba
     fields: dict[str, tuple[type[Any], Any]] = {}
     for param_name, param in sig.parameters.items():
         # This is handled externally
-        if param_name == "cancellation_token" or param_name == "self":
+        if param_name in ["cancellation_token", "self"]:
             continue
 
         if param.annotation is inspect.Parameter.empty:
@@ -196,10 +196,7 @@ def return_type(self) -> type[Any]:
         return self._return_type
 
     def return_value_as_string(self, value: Any) -> str:
-        if isinstance(value, BaseModel):
-            return value.model_dump_json()
-
-        return str(value)
+        return value.model_dump_json() if isinstance(value, BaseModel) else str(value)
 
     @abstractmethod
     def run(self, args: ArgsT | None = None) -> ReturnT: ...
@@ -225,7 +222,7 @@ def __init__(
             if isinstance(func_definition, functools.partial)
             else name or func_definition.__name__
         )
-        args_model = args_base_model_from_signature(func_name + "args", self._signature)
+        args_model = args_base_model_from_signature(f"{func_name}args", self._signature)
         return_type = self._signature.return_annotation
         super().__init__(args_model, return_type, func_name, description, strict)