Skip to content

Commit

Permalink
first pass implementing address reputation action
Browse files Browse the repository at this point in the history
  • Loading branch information
stat committed Jan 17, 2025
1 parent 09f470d commit f070650
Show file tree
Hide file tree
Showing 6 changed files with 318 additions and 0 deletions.
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from cdp_agentkit_core.actions.cdp_action import CdpAction
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
Expand Down Expand Up @@ -31,6 +32,7 @@ def get_all_cdp_actions() -> list[type[CdpAction]]:
__all__ = [
"CDP_ACTIONS",
"CdpAction",
"AddressReputationAction",
"DeployNftAction",
"DeployTokenAction",
"GetBalanceAction",
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
from collections.abc import Callable

from cdp import Address
from pydantic import BaseModel, Field, field_validator
import re

from cdp_agentkit_core.actions import CdpAction

# TODO: ask John what he thinks about standardizing responses to be the json API responses
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-sepolia")
- address: The Ethereum address to check
and returns:
"""


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")
def validate_address(cls, v: str) -> str:
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
93 changes: 93 additions & 0 deletions cdp-agentkit-core/python/tests/actions/test_address_reputation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
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(address_factory):
"""Test successful address reputation check."""
# Create the model and reputation instances
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 MockAddress:
mock_address_instance = MockAddress.return_value
mock_address_instance.reputation.return_value = mock_reputation

action_response = check_address_reputation(MOCK_ADDRESS, MOCK_NETWORK)
expected_response = str(mock_reputation)

MockAddress.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(address_factory):
"""Test address reputation check failure."""
with patch('cdp_agentkit_core.actions.address_reputation.Address') as MockAddress:
mock_address_instance = MockAddress.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"

MockAddress.assert_called_once_with(MOCK_NETWORK, MOCK_ADDRESS)
mock_address_instance.reputation.assert_called_once()
assert action_response == expected_response
56 changes: 56 additions & 0 deletions cdp-agentkit-core/typescript/src/actions/cdp/address_reputation.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
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-sepolia")
- address: The Ethereum address to check
`;

/**
* Input schema for address reputation check.
*/
export const AddressReputationInput = z
.object({
network: z
.string()
.describe("The network to check the address on"),
address: z
.string()
.regex(/^0x[a-fA-F0-9]{40}$/, "Invalid Ethereum address format")
.describe("The Ethereum address to check"),
})
.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.address, args.network);
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;
}
3 changes: 3 additions & 0 deletions cdp-agentkit-core/typescript/src/actions/cdp/index.ts
Original file line number Diff line number Diff line change
@@ -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";
Expand All @@ -21,6 +22,7 @@ import { WOW_ACTIONS } from "./defi/wow";
*/
export function getAllCdpActions(): CdpAction<CdpActionSchemaAny>[] {
return [
new AddressReputationAction(),
new GetWalletDetailsAction(),
new DeployNftAction(),
new DeployTokenAction(),
Expand All @@ -41,6 +43,7 @@ export const CDP_ACTIONS = getAllCdpActions().concat(WOW_ACTIONS);
export {
CdpAction,
CdpActionSchemaAny,
AddressReputationAction,
GetWalletDetailsAction,
DeployNftAction,
DeployTokenAction,
Expand Down
100 changes: 100 additions & 0 deletions cdp-agentkit-core/typescript/src/tests/address_reputation_test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
import { Wallet, Address } from "@coinbase/coinbase-sdk";

Check failure on line 1 in cdp-agentkit-core/typescript/src/tests/address_reputation_test.ts

View workflow job for this annotation

GitHub Actions / lint-cdp-agentkit-core-typescript

'Wallet' is defined but never used
import { AddressReputationAction, AddressReputationInput } from "../actions/cdp/address_reputation";

Check failure on line 2 in cdp-agentkit-core/typescript/src/tests/address_reputation_test.ts

View workflow job for this annotation

GitHub Actions / lint-cdp-agentkit-core-typescript

'AddressReputationInput' is defined but never used

const MOCK_ADDRESS = "0x1234567890123456789012345678901234567890";
const MOCK_NETWORK = "base-sepolia";

jest.mock("@coinbase/coinbase-sdk", () => ({
Address: jest.fn()

Check failure on line 8 in cdp-agentkit-core/typescript/src/tests/address_reputation_test.ts

View workflow job for this annotation

GitHub Actions / lint-cdp-agentkit-core-typescript

Insert `,`
}));

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

Check failure on line 68 in cdp-agentkit-core/typescript/src/tests/address_reputation_test.ts

View workflow job for this annotation

GitHub Actions / lint-cdp-agentkit-core-typescript

Insert `,`
},
toString: () => "Address Reputation: (score=85, metadata=(...))"

Check failure on line 70 in cdp-agentkit-core/typescript/src/tests/address_reputation_test.ts

View workflow job for this annotation

GitHub Actions / lint-cdp-agentkit-core-typescript

Insert `,`
};

mockAddress.reputation.mockResolvedValue(mockReputation as any);

Check failure on line 73 in cdp-agentkit-core/typescript/src/tests/address_reputation_test.ts

View workflow job for this annotation

GitHub Actions / lint-cdp-agentkit-core-typescript

Unexpected any. Specify a different type

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}`);
});
});

0 comments on commit f070650

Please sign in to comment.