Fix lint issues

cdp-agentkit-core/cdp_agentkit_core/actions/__init__.py

@@ -8,10 +8,10 @@
 from cdp_agentkit_core.actions.request_faucet_funds import RequestFaucetFundsAction
 from cdp_agentkit_core.actions.trade import TradeAction
 from cdp_agentkit_core.actions.transfer import TransferAction
+from cdp_agentkit_core.actions.webhook import CreateWebhookAction
 from cdp_agentkit_core.actions.wow.buy_token import WowBuyTokenAction
 from cdp_agentkit_core.actions.wow.create_token import WowCreateTokenAction
 from cdp_agentkit_core.actions.wow.sell_token import WowSellTokenAction
-from cdp_agentkit_core.actions.webhook import CreateWebhookAction
 
 
 # WARNING: All new CdpAction subclasses must be imported above, otherwise they will not be discovered
@@ -29,6 +29,7 @@ def get_all_cdp_actions() -> list[type[CdpAction]]:
 __all__ = [
     "CDP_ACTIONS",
     "CdpAction",
+    "CreateWebhookAction",
     "DeployNftAction",
     "DeployTokenAction",
     "GetBalanceAction",
@@ -41,5 +42,4 @@ def get_all_cdp_actions() -> list[type[CdpAction]]:
     "WowBuyTokenAction",
     "WowCreateTokenAction",
     "WowSellTokenAction",
-    "CreateWebhookAction",
 ]

cdp-agentkit-core/cdp_agentkit_core/actions/webhook.py

@@ -1,16 +1,16 @@
 from enum import Enum
-from typing import Any, Dict, List, Optional, Union
+from typing import Any
 
-from cdp import Wallet, Webhook
-from pydantic import BaseModel, Field, HttpUrl, model_validator, field_validator
-
-from cdp_agentkit_core.actions import CdpAction
+from cdp import Webhook
 from cdp.client.models.webhook import WebhookEventTypeFilter
-from cdp.client.models.webhook_wallet_activity_filter import WebhookWalletActivityFilter
 from cdp.client.models.webhook_smart_contract_event_filter import WebhookSmartContractEventFilter
+from cdp.client.models.webhook_wallet_activity_filter import WebhookWalletActivityFilter
+from pydantic import BaseModel, Field, HttpUrl, field_validator, model_validator
+
+from cdp_agentkit_core.actions import CdpAction
 
 CREATE_WEBHOOK_PROMPT = """
-Create a new webhook to receive real-time updates for on-chain events. 
+Create a new webhook to receive real-time updates for on-chain events.
 Supports monitoring wallet activity or smart contract events by specifying:
 - Callback URL for receiving events
 - Event type (wallet_activity, smart_contract_event_activity, erc20_transfer or erc721_transfer)
@@ -24,21 +24,24 @@
 
 class WebhookEventType(str, Enum):
     """Valid webhook event types."""
+
     WALLET_ACTIVITY = "wallet_activity"
     SMART_CONTRACT_EVENT_ACTIVITY = "smart_contract_event_activity"
     ERC20_TRANSFER = "erc20_transfer"
     ERC721_TRANSFER = "erc721_transfer"
 
 class WebhookNetworks(str, Enum):
     """Networks available for creating webhooks."""
+
     BASE_MAINNET = "base-mainnet"
     BASE_SEPOLIA = "base-sepolia"
 
 class EventFilter(BaseModel):
     """Schema for event filters."""
-    from_address: Optional[str] = Field(None, description="Sender address for token transfers")
-    to_address: Optional[str] = Field(None, description="Recipient address for token transfers")
-    contract_address: Optional[str] = Field(None, description="Contract address for token transfers")
+
+    from_address: str | None = Field(None, description="Sender address for token transfers")
+    to_address: str | None = Field(None, description="Recipient address for token transfers")
+    contract_address: str | None = Field(None, description="Contract address for token transfers")
 
     @model_validator(mode='after')
     def validate_at_least_one_filter(self) -> 'EventFilter':
@@ -49,22 +52,24 @@ def validate_at_least_one_filter(self) -> 'EventFilter':
 
 class EventTypeFilter(BaseModel):
     """Schema for event type filter."""
-    addresses: Optional[List[str]] = Field(None, description="List of wallet or contract addresses to monitor")
+
+    addresses: list[str] | None = Field(None, description="List of wallet or contract addresses to monitor")
 
     @field_validator('addresses')
     @classmethod
-    def validate_addresses_not_empty(cls, v: Optional[List[str]]) -> Optional[List[str]]:
+    def validate_addresses_not_empty(cls, v: list[str] | None) -> list[str] | None:
         """Ensure addresses list is not empty when provided."""
         if v is not None and len(v) == 0:
             raise ValueError("addresses must contain at least one value when provided")
         return v
 
 class WebhookInput(BaseModel):
     """Input schema for create webhook action."""
+
     notification_uri: HttpUrl = Field(..., description="The callback URL where webhook events will be sent")
     event_type: WebhookEventType
-    event_type_filter: Optional[EventTypeFilter] = None
-    event_filters: Optional[List[EventFilter]] = None
+    event_type_filter: EventTypeFilter | None = None
+    event_filters: list[EventFilter] | None = None
     network_id: WebhookNetworks
 
     @model_validator(mode='after')
@@ -95,13 +100,13 @@ def validate_filters(self) -> 'WebhookInput':
         return self
 
 def create_webhook(
-    notification_uri: Union[str, HttpUrl],
+    notification_uri: str | HttpUrl,
     event_type: str,
     network_id: str,
-    event_type_filter: Optional[Dict[str, Any]] = None,
-    event_filters: Optional[List[Dict[str, Any]]] = None,
+    event_type_filter: dict[str, Any] | None = None,
+    event_filters: list[dict[str, Any]] | None = None,
 ) -> str:
-    """Creates a new webhook for monitoring on-chain events.
+    """Create a new webhook for monitoring on-chain events.
 
     Args:
         notification_uri: The callback URL where webhook events will be sent
@@ -112,8 +117,8 @@ def create_webhook(
 
     Returns:
         str: Details of the created webhook
-    """
 
+    """
     print(f"notification_uri: {notification_uri}")
     print(f"event_type_filter: {event_type_filter}")
     print(f"event_filters: {event_filters}")
@@ -155,12 +160,13 @@ def create_webhook(
         print(f"webhook_options: {webhook_options}")
         webhook = Webhook.create(**webhook_options)
         return f"The webhook was successfully created: {webhook}\n\n"
-    
+
     except Exception as error:
-        return f"Error: {str(error)}"
+        return f"Error: {error!s}"
 
 class CreateWebhookAction(CdpAction):
     """Create webhook action."""
+
     name: str = "create_webhook"
     description: str = CREATE_WEBHOOK_PROMPT
     args_schema: type[BaseModel] = WebhookInput

cdp-agentkit-core/tests/actions/test_webhook.py

@@ -1,5 +1,6 @@
+from unittest.mock import Mock, patch
+
 import pytest
-from unittest.mock import patch, Mock
 
 from cdp_agentkit_core.actions.webhook import (
     WebhookInput,
@@ -15,13 +16,14 @@
 
 @pytest.fixture
 def mock_webhook():
+    """Provide a mocked Webhook instance for testing."""
     with patch('cdp_agentkit_core.actions.webhook.Webhook') as mock:
         mock_instance = Mock()
         mock.create.return_value = mock_instance
         yield mock
 
 def test_webhook_input_valid_parsing():
-    """Test successful parsing of valid webhook inputs"""
+    """Test successful parsing of valid webhook inputs."""
     # Test wallet activity webhook input
     valid_input = {
         "notification_uri": MOCK_URL,
@@ -31,7 +33,7 @@ def test_webhook_input_valid_parsing():
         },
         "network_id": MOCK_NETWORK
     }
-    
+
     result = WebhookInput.model_validate(valid_input)
     assert str(result.notification_uri) == MOCK_URL
     assert result.event_type == MOCK_EVENT_TYPE
@@ -47,20 +49,20 @@ def test_webhook_input_valid_parsing():
         }],
         "network_id": MOCK_NETWORK
     }
-    
+
     result = WebhookInput.model_validate(another_valid_input)
     assert str(result.notification_uri) == MOCK_URL
     assert result.event_type == "erc721_transfer"
     assert result.event_filters[0].from_address == MOCK_ADDRESS
 
 def test_webhook_input_invalid_parsing():
-    """Test parsing failure for invalid webhook input"""
+    """Test parsing failure for invalid webhook input."""
     empty_input = {}
     with pytest.raises(ValueError):
         WebhookInput.model_validate(empty_input)
 
 def test_create_wallet_activity_webhook(mock_webhook):
-    """Test creating wallet activity webhook"""
+    """Test creating wallet activity webhook."""
     args = {
         "notification_uri": MOCK_URL,
         "event_type": MOCK_EVENT_TYPE,
@@ -69,14 +71,14 @@ def test_create_wallet_activity_webhook(mock_webhook):
         },
         "network_id": MOCK_NETWORK
     }
-    
+
     response = create_webhook(**args)
-    
+
     assert mock_webhook.create.call_count == 1
     assert SUCCESS_MESSAGE in response
 
 def test_create_smart_contract_activity_webhook(mock_webhook):
-    """Test creating smart contract activity webhook"""
+    """Test creating smart contract activity webhook."""
     args = {
         "notification_uri": MOCK_URL,
         "event_type": "smart_contract_event_activity",
@@ -85,14 +87,14 @@ def test_create_smart_contract_activity_webhook(mock_webhook):
         },
         "network_id": MOCK_NETWORK
     }
-    
+
     response = create_webhook(**args)
-    
+
     assert mock_webhook.create.call_count == 1
     assert SUCCESS_MESSAGE in response
 
 def test_create_erc20_transfer_webhook(mock_webhook):
-    """Test creating ERC20 transfer webhook"""
+    """Test creating ERC20 transfer webhook."""
     args = {
         "notification_uri": MOCK_URL,
         "event_type": "erc20_transfer",
@@ -104,14 +106,14 @@ def test_create_erc20_transfer_webhook(mock_webhook):
         }],
         "network_id": MOCK_NETWORK
     }
-    
+
     response = create_webhook(**args)
-    
+
     assert mock_webhook.create.call_count == 1
     assert SUCCESS_MESSAGE in response
 
 def test_create_erc721_transfer_webhook(mock_webhook):
-    """Test creating ERC721 transfer webhook"""
+    """Test creating ERC721 transfer webhook."""
     args = {
         "notification_uri": MOCK_URL,
         "event_type": "erc721_transfer",
@@ -120,17 +122,17 @@ def test_create_erc721_transfer_webhook(mock_webhook):
         }],
         "network_id": MOCK_NETWORK
     }
-    
+
     response = create_webhook(**args)
-    
+
     assert mock_webhook.create.call_count == 1
     assert SUCCESS_MESSAGE in response
 
 def test_create_webhook_error_handling(mock_webhook):
-    """Test error handling when creating webhook fails"""
+    """Test error handling when creating webhook fails."""
     error_msg = "Failed to create webhook"
     mock_webhook.create.side_effect = Exception(error_msg)
-    
+
     args = {
         "notification_uri": MOCK_URL,
         "event_type": MOCK_EVENT_TYPE,
@@ -139,8 +141,8 @@ def test_create_webhook_error_handling(mock_webhook):
         },
         "network_id": MOCK_NETWORK
     }
-    
+
     response = create_webhook(**args)
-    
+
     assert mock_webhook.create.call_count == 1
     assert f"Error: {error_msg}" in response