The UserPersistenceManager manages user identities using wallet addresses as primary identifiers. All user actions are verified using wallet signatures, ensuring secure and authenticated user operations.
File: tools/user_persistence_manager.py
Class: UserPersistenceManager
Version: 1.0.0
Status: ✅ Active
class UserPersistenceManager:
- participants_dir: Path - Participants directory
- users: Dict[str, UserIdentity] - User registry
- signatures: Dict[str, List] - Signature history
- user_agents: Dict[str, List] - User agents
@dataclass
class UserIdentity:
wallet_address: str # Primary identifier
user_id: str # Generated user ID
created_at: float # Creation timestamp
last_active: float # Last activity timestamp
signature_count: int # Total signatures
agent_count: int # Number of agents
metadata: Dict[str, Any] # Additional metadata
public_key: Optional[str] # Public key
@dataclass
class SignatureVerification:
is_valid: bool # Verification result
message: str # Signed message
signature: str # Signature
recovered_address: Optional[str] # Recovered address
timestamp: float # Verification timestamp
register_userRegisters a new user with signature verification.
Parameters:
wallet_address (str): User's wallet addresssignature (str): Signature of challenge messagemessage (str): Challenge messagemetadata (Dict, optional): User metadataReturns:
Tuple[bool, str] # (success, message)
verify_signatureVerifies a wallet signature.
Parameters:
wallet_address (str): Wallet addressmessage (str): Signed messagesignature (str): SignatureReturns: SignatureVerification object
verify_user_actionVerifies a user action with signature.
Parameters:
wallet_address (str): User walletaction (str): Action namesignature (str): Signaturemessage (str): Challenge messageReturns:
Tuple[bool, str] # (success, message)
create_user_agentCreates a user agent with signature verification.
Parameters:
wallet_address (str): User walletagent_id (str): Agent identifieragent_type (str): Agent typesignature (str): Signaturemessage (str): Challenge messagemetadata (Dict, optional): Agent metadataReturns:
Tuple[bool, str, Optional[str]] # (success, message, agent_wallet)
delete_user_agentDeletes a user agent with signature verification.
Parameters:
wallet_address (str): User walletagent_id (str): Agent to deletesignature (str): Signaturemessage (str): Challenge messageReturns:
Tuple[bool, str] # (success, message)
get_user_agentsGets all agents for a user.
Parameters:
wallet_address (str): User walletReturns: List[Dict[str, Any]]
get_user_statsGets user statistics.
Parameters:
wallet_address (str): User walletReturns: Dict[str, Any]
generate_challenge_messageGenerates a challenge message for signature.
Parameters:
wallet_address (str): User walletaction (str): Action nameReturns: str (challenge message)
from tools.user_persistence_manager import UserPersistenceManager
manager = UserPersistenceManager()
Generate challenge
challenge = manager.generate_challenge_message(
wallet_address="0x...",
action="register"
)
User signs challenge, then register
success, message = manager.register_user(
wallet_address="0x...",
signature="0x...",
message=challenge,
metadata={"name": "User Name"}
)
# Generate challenge
challenge = manager.generate_challenge_message(
wallet_address="0x...",
action="create_agent"
)
Create agent
success, message, agent_wallet = manager.create_user_agent(
wallet_address="0x...",
agent_id="user_agent_001",
agent_type="analysis_agent",
signature="0x...",
message=challenge
)
# Verify user action
success, message = manager.verify_user_action(
wallet_address="0x...",
action="execute_task",
signature="0x...",
message=challenge
)
data/participants/users.json - User identitiesdata/participants/signatures.json - Signature historydata/participants/user_agents.json - User agentsdata/participants/
├── users.json
├── signatures.json
└── user_agents.json
All operations require:
Challenge messages include:
Tracks all signatures:
Uses eth_account for:
# 1. Register user
challenge = manager.generate_challenge_message("0x...", "register")
success, _ = manager.register_user("0x...", signature, challenge)
2. Create agent
challenge = manager.generate_challenge_message("0x...", "create_agent")
success, _, agent_wallet = manager.create_user_agent(
"0x...", "agent_001", "analysis", signature, challenge
)
3. Get user stats
stats = manager.get_user_stats("0x...")
eth_account: Ethereum account managementdataclasses: Data structureshashlib: Nonce generationutils.config.Config: ConfigurationUses Ethereum message signing:
message_hash = encode_defunct(text=message)
recovered_address = Account.recover_message(message_hash, signature=signature)