Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

NFT functions for general agent #582

Merged
merged 7 commits into from
Dec 6, 2024
Merged
Show file tree
Hide file tree
Changes from 6 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11,034 changes: 0 additions & 11,034 deletions poetry.lock

Large diffs are not rendered by default.

Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@ class SendPaidMessageToAnotherAgent(Function):
@property
def description(self) -> str:
return f"""Use {SendPaidMessageToAnotherAgent.__name__} to send a message to an another agent, given his wallet address.
Fee for sending the message is {TRANSACTION_MESSAGE_FEE} xDai."""

@property
def example_args(self) -> list[str]:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
from prediction_market_agent.agents.microchain_agent.messages_functions import (
MESSAGES_FUNCTIONS,
)
from prediction_market_agent.agents.microchain_agent.nft_functions import NFT_FUNCTIONS
from prediction_market_agent.agents.microchain_agent.omen_functions import (
OMEN_FUNCTIONS,
)
Expand Down Expand Up @@ -156,6 +157,9 @@ def build_agent_functions(
if functions_config.include_messages_functions:
functions.extend(f() for f in MESSAGES_FUNCTIONS)

if functions_config.include_nft_functions:
functions.extend(f() for f in NFT_FUNCTIONS)

if long_term_memory:
functions.append(
RememberPastActions(long_term_memory=long_term_memory, model=model)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@ class MicrochainAgentKeys(APIKeys):
ENABLE_SOCIAL_MEDIA: bool = False
# Double check to not spend big money during testing.
SENDING_XDAI_CAP: float | None = OMEN_TINY_BET_AMOUNT
# Double check to not transfer NFTs during testing.
ENABLE_NFT_TRANSFER: bool = False

def cap_sending_xdai(self, amount: xDai) -> xDai:
if self.SENDING_XDAI_CAP is None:
Expand Down
97 changes: 97 additions & 0 deletions prediction_market_agent/agents/microchain_agent/nft_functions.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
from microchain import Function
from prediction_market_agent_tooling.loggers import logger
from prediction_market_agent_tooling.tools.contract import (
ContractOwnableERC721OnGnosisChain,
)
from web3 import Web3

from prediction_market_agent.agents.microchain_agent.microchain_agent_keys import (
MicrochainAgentKeys,
)


class BalanceOfNFT(Function):
@property
def description(self) -> str:
return "Returns the number of given NFT owned by the specified address."

@property
def example_args(self) -> list[str]:
return [
"0xNFTAddress",
"0xOwnerddress",
]

def __call__(
self,
nft_address: str,
owner_address: str,
) -> int:
contract = ContractOwnableERC721OnGnosisChain(
address=Web3.to_checksum_address(nft_address)
)
balance: int = contract.balanceOf(Web3.to_checksum_address(owner_address))
return balance
Comment on lines +27 to +34
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Add Input Validation and Exception Handling

When interacting with smart contracts, it's important to validate inputs and handle potential exceptions to prevent unexpected failures.

Consider adding input validation and exception handling:

def __call__(
    self,
    nft_address: str,
    owner_address: str,
) -> int:
    try:
        contract = ContractOwnableERC721OnGnosisChain(
            address=Web3.to_checksum_address(nft_address)
        )
        balance: int = contract.balanceOf(Web3.to_checksum_address(owner_address))
        return balance
    except Exception as e:
        logger.error(f"Error fetching NFT balance: {e}")
        raise ValueError("Failed to retrieve NFT balance.")



class OwnerOfNFT(Function):
@property
def description(self) -> str:
return "Returns the owner address of the specified NFT token ID."

@property
def example_args(self) -> list[str]:
return ["0xNFTAddress", "1"]

def __call__(
self,
nft_address: str,
token_id: int,
) -> str:
contract = ContractOwnableERC721OnGnosisChain(
address=Web3.to_checksum_address(nft_address)
)
owner_address = contract.ownerOf(token_id)
return owner_address
Comment on lines +51 to +55
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Add Exception Handling for Smart Contract Calls

Similar to the BalanceOfNFT class, add exception handling in the OwnerOfNFT class to manage potential errors when calling ownerOf.

def __call__(
    self,
    nft_address: str,
    token_id: int,
) -> str:
    try:
        contract = ContractOwnableERC721OnGnosisChain(
            address=Web3.to_checksum_address(nft_address)
        )
        owner_address = contract.ownerOf(token_id)
        return owner_address
    except Exception as e:
        logger.error(f"Error fetching NFT owner: {e}")
        raise ValueError("Failed to retrieve NFT owner.")



class SafeTransferFromNFT(Function):
@property
def description(self) -> str:
return "Transfers the specified NFT token ID from one address to another."

@property
def example_args(self) -> list[str]:
return [
"0xNFTAddress",
"0xRecipientAddress",
"1",
]

def __call__(
self,
nft_address: str,
to_address: str,
token_id: int,
) -> str:
keys = MicrochainAgentKeys()
contract = ContractOwnableERC721OnGnosisChain(
address=Web3.to_checksum_address(nft_address)
)
if keys.ENABLE_NFT_TRANSFER:
contract.safeTransferFrom(
api_keys=keys,
from_address=keys.bet_from_address,
to_address=Web3.to_checksum_address(to_address),
tokenId=token_id,
)
else:
logger.warning("NFT transfer is disabled in the environment.")
return "Token transferred successfully."
Comment on lines +77 to +90
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Return Appropriate Message Based on Transfer Outcome

Currently, the method returns "Token transferred successfully." even when NFT transfers are disabled, which might be misleading.

Adjust the return statements to reflect the actual outcome:

def __call__(
    self,
    nft_address: str,
    to_address: str,
    token_id: int,
) -> str:
    keys = MicrochainAgentKeys()
    contract = ContractOwnableERC721OnGnosisChain(
        address=Web3.to_checksum_address(nft_address)
    )
    if keys.ENABLE_NFT_TRANSFER:
        contract.safeTransferFrom(
            api_keys=keys,
            from_address=keys.bet_from_address,
            to_address=Web3.to_checksum_address(to_address),
            tokenId=token_id,
        )
        return "Token transferred successfully."
    else:
        logger.warning("NFT transfer is disabled in the environment.")
        return "NFT transfer is disabled."


Comment on lines +77 to +91
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Handle Exceptions During NFT Transfer

When performing the NFT transfer, exceptions may occur due to issues like insufficient permissions or network errors.

Add exception handling to manage potential errors:

def __call__(
    self,
    nft_address: str,
    to_address: str,
    token_id: int,
) -> str:
    keys = MicrochainAgentKeys()
    contract = ContractOwnableERC721OnGnosisChain(
        address=Web3.to_checksum_address(nft_address)
    )
    if keys.ENABLE_NFT_TRANSFER:
        try:
            contract.safeTransferFrom(
                api_keys=keys,
                from_address=keys.bet_from_address,
                to_address=Web3.to_checksum_address(to_address),
                tokenId=token_id,
            )
            return "Token transferred successfully."
        except Exception as e:
            logger.error(f"Error transferring NFT: {e}")
            raise ValueError("Failed to transfer NFT.")
    else:
        logger.warning("NFT transfer is disabled in the environment.")
        return "NFT transfer is disabled."


NFT_FUNCTIONS: list[type[Function]] = [
BalanceOfNFT,
OwnerOfNFT,
SafeTransferFromNFT,
]
4 changes: 4 additions & 0 deletions prediction_market_agent/agents/microchain_agent/prompts.py
Original file line number Diff line number Diff line change
Expand Up @@ -136,6 +136,7 @@ class FunctionsConfig(BaseModel):
include_sending_functions: bool
include_twitter_functions: bool
include_messages_functions: bool
include_nft_functions: bool
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Set Default Value for include_nft_functions

In the FunctionsConfig class, include_nft_functions is declared without a default value, which could lead to initialization errors.

Assign a default value to ensure proper instantiation:

class FunctionsConfig(BaseModel):
    # ...
    include_messages_functions: bool
    include_nft_functions: bool = False


@staticmethod
def from_system_prompt_choice(
Expand All @@ -149,6 +150,7 @@ def from_system_prompt_choice(
include_sending_functions = False
include_twitter_functions = False
include_messages_functions = False
include_nft_functions = False

if system_prompt_choice == SystemPromptChoice.JUST_BORN:
include_learning_functions = True
Expand All @@ -171,6 +173,7 @@ def from_system_prompt_choice(

elif system_prompt_choice == SystemPromptChoice.DARE_YOU_GET_MY_RESOURCES_AGENT:
include_messages_functions = True
include_nft_functions = True

return FunctionsConfig(
include_trading_functions=include_trading_functions,
Expand All @@ -181,6 +184,7 @@ def from_system_prompt_choice(
include_sending_functions=include_sending_functions,
include_twitter_functions=include_twitter_functions,
include_messages_functions=include_messages_functions,
include_nft_functions=include_nft_functions,
)


Expand Down
Loading