feat: address reputation (#127)

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

@@ -1,4 +1,6 @@
-from cdp_agentkit_core.actions.cdp_action import CdpAction
+from cdp_agentkit_core.actions.cdp_action import CdpAction  # noqa: I001
+
+from cdp_agentkit_core.actions.address_reputation import AddressReputationAction
 from cdp_agentkit_core.actions.deploy_nft import DeployNftAction
 from cdp_agentkit_core.actions.deploy_token import DeployTokenAction
 from cdp_agentkit_core.actions.get_balance import GetBalanceAction
@@ -38,6 +40,7 @@ def get_all_cdp_actions() -> list[type[CdpAction]]:
 __all__ = [
     "CDP_ACTIONS",
     "CdpAction",
+    "AddressReputationAction",
     "DeployNftAction",
     "DeployTokenAction",
     "GetBalanceAction",

cdp-agentkit-core/python/cdp_agentkit_core/actions/address_reputation.py

@@ -0,0 +1,72 @@
+import re
+from collections.abc import Callable
+
+from cdp import Address
+from pydantic import BaseModel, Field, field_validator
+
+from cdp_agentkit_core.actions import CdpAction
+
+ADDRESS_REPUTATION_PROMPT = """
+This tool checks the reputation of an address on a given network. It takes:
+
+- network: The network the address is on (e.g. "base-mainnet")
+- address: The Ethereum address to check
+
+Important notes:
+- This tool will not work on base-sepolia, you can default to using base-mainnet instead
+- The wallet's default address and its network may be used if not provided
+"""
+
+
+class AddressReputationInput(BaseModel):
+    """Input argument schema for checking address reputation."""
+
+    address: str = Field(..., description="The Ethereum address to check")
+    network: str = Field(..., description="The network to check the address on")
+
+    @field_validator("address")
+    @classmethod
+    def validate_address(cls, v: str) -> str:
+        """Validate that the provided address is a valid Ethereum address.
+
+        Args:
+            v (str): The address string to validate
+
+        Returns:
+            str: The validated address string
+
+        Raises:
+            ValueError: If the address format is invalid
+
+        """
+        if not re.match(r"^0x[a-fA-F0-9]{40}$", v):
+            raise ValueError("Invalid Ethereum address format")
+        return v
+
+
+def check_address_reputation(address: str, network: str) -> str:
+    """Check the reputation of an address.
+
+    Args:
+        address (str): The Ethereum address to check
+        network (str): The network the address is on
+
+    Returns:
+        str: A string containing the reputation json data or error message
+
+    """
+    try:
+        address = Address(network, address)
+        reputation = address.reputation()
+        return str(reputation)
+    except Exception as e:
+        return f"Error checking address reputation: {e!s}"
+
+
+class AddressReputationAction(CdpAction):
+    """Address reputation check action."""
+
+    name: str = "address_reputation"
+    description: str = ADDRESS_REPUTATION_PROMPT
+    args_schema: type[BaseModel] | None = AddressReputationInput
+    func: Callable[..., str] = check_address_reputation

cdp-agentkit-core/python/tests/actions/test_address_reputation.py

@@ -0,0 +1,96 @@
+from unittest.mock import patch
+
+import pytest
+from cdp.address_reputation import (
+    AddressReputation,
+    AddressReputationMetadata,
+    AddressReputationModel,
+)
+
+from cdp_agentkit_core.actions.address_reputation import (
+    AddressReputationAction,
+    AddressReputationInput,
+    check_address_reputation,
+)
+
+MOCK_ADDRESS = "0x1234567890123456789012345678901234567890"
+MOCK_NETWORK = "base-sepolia"
+
+
+def test_address_reputation_action_initialization():
+    """Test AddressReputationAction initialization and attributes."""
+    action = AddressReputationAction()
+
+    assert action.name == "address_reputation"
+    assert action.args_schema == AddressReputationInput
+    assert callable(action.func)
+
+
+def test_address_reputation_input_model_valid():
+    """Test AddressReputationInput accepts valid parameters."""
+    valid_input = AddressReputationInput(
+        network=MOCK_NETWORK,
+        address=MOCK_ADDRESS,
+    )
+    assert valid_input.network == MOCK_NETWORK
+    assert valid_input.address == MOCK_ADDRESS
+
+
+def test_address_reputation_input_model_missing_params():
+    """Test AddressReputationInput raises error when params are missing."""
+    with pytest.raises(ValueError):
+        AddressReputationInput()
+
+
+def test_address_reputation_input_model_invalid_address():
+    """Test AddressReputationInput raises error with invalid address format."""
+    with pytest.raises(ValueError, match="Invalid Ethereum address format"):
+        AddressReputationInput(
+            network=MOCK_NETWORK,
+            address="not_an_address"
+        )
+
+
+def test_address_reputation_success():
+    """Test successful address reputation check."""
+    mock_model = AddressReputationModel(
+        score=85,
+        metadata=AddressReputationMetadata(
+            total_transactions=150,
+            unique_days_active=30,
+            longest_active_streak=10,
+            current_active_streak=5,
+            activity_period_days=45,
+            token_swaps_performed=20,
+            bridge_transactions_performed=5,
+            lend_borrow_stake_transactions=10,
+            ens_contract_interactions=2,
+            smart_contract_deployments=1
+        )
+    )
+    mock_reputation = AddressReputation(model=mock_model)
+
+    with patch('cdp_agentkit_core.actions.address_reputation.Address') as mock_address:
+        mock_address_instance = mock_address.return_value
+        mock_address_instance.reputation.return_value = mock_reputation
+
+        action_response = check_address_reputation(MOCK_ADDRESS, MOCK_NETWORK)
+        expected_response = str(mock_reputation)
+
+        mock_address.assert_called_once_with(MOCK_NETWORK, MOCK_ADDRESS)
+        mock_address_instance.reputation.assert_called_once()
+        assert action_response == expected_response
+
+
+def test_address_reputation_failure():
+    """Test address reputation check failure."""
+    with patch('cdp_agentkit_core.actions.address_reputation.Address') as mock_address:
+        mock_address_instance = mock_address.return_value
+        mock_address_instance.reputation.side_effect = Exception("API error")
+
+        action_response = check_address_reputation(MOCK_ADDRESS, MOCK_NETWORK)
+        expected_response = "Error checking address reputation: API error"
+
+        mock_address.assert_called_once_with(MOCK_NETWORK, MOCK_ADDRESS)
+        mock_address_instance.reputation.assert_called_once()
+        assert action_response == expected_response

cdp-agentkit-core/typescript/src/actions/cdp/address_reputation.ts

@@ -0,0 +1,58 @@
+import { Wallet, Address } from "@coinbase/coinbase-sdk";
+import { z } from "zod";
+
+import { CdpAction } from "./cdp_action";
+
+const ADDRESS_REPUTATION_PROMPT = `
+This tool checks the reputation of an address on a given network. It takes:
+
+- network: The network to check the address on (e.g. "base-mainnet")
+- address: The Ethereum address to check
+
+Important notes:
+- This tool will not work on base-sepolia, you can default to using base-mainnet instead
+- The wallet's default address and its network may be used if not provided
+`;
+
+/**
+ * Input schema for address reputation check.
+ */
+export const AddressReputationInput = z
+  .object({
+    address: z
+      .string()
+      .regex(/^0x[a-fA-F0-9]{40}$/, "Invalid Ethereum address format")
+      .describe("The Ethereum address to check"),
+    network: z.string().describe("The network to check the address on"),
+  })
+  .strip()
+  .describe("Input schema for address reputation check");
+
+/**
+ * Check the reputation of an address.
+ *
+ * @param wallet - The wallet instance
+ * @param args - The input arguments for the action
+ * @returns A string containing reputation data or error message
+ */
+export async function checkAddressReputation(
+  args: z.infer<typeof AddressReputationInput>,
+): Promise<string> {
+  try {
+    const address = new Address(args.network, args.address);
+    const reputation = await address.reputation();
+    return reputation.toString();
+  } catch (error) {
+    return `Error checking address reputation: ${error}`;
+  }
+}
+
+/**
+ * Address reputation check action.
+ */
+export class AddressReputationAction implements CdpAction<typeof AddressReputationInput> {
+  public name = "address_reputation";
+  public description = ADDRESS_REPUTATION_PROMPT;
+  public argsSchema = AddressReputationInput;
+  public func = checkAddressReputation;
+}

cdp-agentkit-core/typescript/src/actions/cdp/index.ts

@@ -1,4 +1,5 @@
 import { CdpAction, CdpActionSchemaAny } from "./cdp_action";
+import { AddressReputationAction } from "./address_reputation";
 import { DeployNftAction } from "./deploy_nft";
 import { DeployTokenAction } from "./deploy_token";
 import { GetBalanceAction } from "./get_balance";
@@ -24,6 +25,7 @@ import { WOW_ACTIONS } from "./defi/wow";
  */
 export function getAllCdpActions(): CdpAction<CdpActionSchemaAny>[] {
   return [
+    new AddressReputationAction(),
     new GetWalletDetailsAction(),
     new DeployNftAction(),
     new DeployTokenAction(),
@@ -47,6 +49,7 @@ export const CDP_ACTIONS = getAllCdpActions()
 export {
   CdpAction,
   CdpActionSchemaAny,
+  AddressReputationAction,
   GetWalletDetailsAction,
   DeployNftAction,
   DeployTokenAction,

cdp-agentkit-core/typescript/src/tests/address_reputation_test.ts

@@ -0,0 +1,102 @@
+import { Address } from "@coinbase/coinbase-sdk";
+import { AddressReputationAction } from "../actions/cdp/address_reputation";
+
+const MOCK_ADDRESS = "0x1234567890123456789012345678901234567890";
+const MOCK_NETWORK = "base-sepolia";
+
+jest.mock("@coinbase/coinbase-sdk", () => ({
+  Address: jest.fn(),
+}));
+
+describe("Address Reputation Input", () => {
+  const action = new AddressReputationAction();
+
+  it("should successfully parse valid input", () => {
+    const validInput = {
+      network: MOCK_NETWORK,
+      address: MOCK_ADDRESS,
+    };
+
+    const result = action.argsSchema.safeParse(validInput);
+
+    expect(result.success).toBe(true);
+    expect(result.data).toEqual(validInput);
+  });
+
+  it("should fail parsing empty input", () => {
+    const emptyInput = {};
+    const result = action.argsSchema.safeParse(emptyInput);
+
+    expect(result.success).toBe(false);
+  });
+
+  it("should fail with invalid address", () => {
+    const invalidInput = {
+      network: MOCK_NETWORK,
+      address: "not_an_address",
+    };
+    const result = action.argsSchema.safeParse(invalidInput);
+
+    expect(result.success).toBe(false);
+  });
+});
+
+describe("Address Reputation Action", () => {
+  let mockAddress: jest.Mocked<Address>;
+
+  beforeEach(() => {
+    mockAddress = {
+      reputation: jest.fn(),
+    } as unknown as jest.Mocked<Address>;
+
+    (Address as unknown as jest.Mock).mockImplementation(() => mockAddress);
+  });
+
+  it("should successfully check address reputation", async () => {
+    const mockReputation = {
+      score: 85,
+      metadata: {
+        total_transactions: 150,
+        unique_days_active: 30,
+        longest_active_streak: 10,
+        current_active_streak: 5,
+        activity_period_days: 45,
+        token_swaps_performed: 20,
+        bridge_transactions_performed: 5,
+        lend_borrow_stake_transactions: 10,
+        ens_contract_interactions: 2,
+        smart_contract_deployments: 1,
+      },
+
+      // TODO: remove this once AddressReputation is exported from the sdk
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    } as unknown as jest.Mocked<any>;
+
+    mockAddress.reputation.mockResolvedValue(mockReputation);
+
+    const args = {
+      network: MOCK_NETWORK,
+      address: MOCK_ADDRESS,
+    };
+
+    const action = new AddressReputationAction();
+    const response = await action.func(args);
+
+    expect(response).toBe(mockReputation.toString());
+  });
+
+  it("should handle errors gracefully", async () => {
+    const error = new Error("API error");
+    mockAddress.reputation.mockRejectedValue(error);
+
+    const args = {
+      network: MOCK_NETWORK,
+      address: MOCK_ADDRESS,
+    };
+
+    const action = new AddressReputationAction();
+    const response = await action.func(args);
+
+    expect(response).toBe(`Error checking address reputation: ${error}`);
+  });
+});