"""
Tools for Keet messaging integration.
This module provides tools for interacting with the Keet messaging platform.
"""
import asyncio
import logging
from typing import Dict, List, Any, Optional, Union
from pathlib import Path
from ollamatools.plugins.base import ToolPlugin
from .client import KeetClient
from .utils import extract_room_key_from_invite, is_valid_hypercore_key, normalize_hypercore_key
logger = logging.getLogger(__name__)
class KeetTools(ToolPlugin):
"""
Tools for interacting with the Keet messaging platform.
"""
def __init__(self, storage_path: Optional[Union[str, Path]] = None):
"""
Initialize the Keet tools.
Args:
storage_path: Path to store Keet data. If None, a default path will be used.
"""
super().__init__()
self.client = KeetClient(storage_path)
logger.info("Initialized Keet tools")
async def get_identity(self) -> Dict[str, Any]:
"""
Get the current identity or create a new one if none exists.
Returns:
Dict[str, Any]: The identity information
"""
return await self.client.get_identity()
async def create_identity(self, name: Optional[str] = None) -> Dict[str, Any]:
"""
Create a new identity.
Args:
name: The display name for the identity
Returns:
Dict[str, Any]: The created identity information
"""
return await self.client.create_identity(name)
async def get_recent_rooms(self) -> List[Dict[str, Any]]:
"""
Get a list of recent rooms.
Returns:
List[Dict[str, Any]]: List of room information
"""
return await self.client.get_recent_rooms()
async def join_room(self, room_key: str) -> bool:
"""
Join a room using a room key or invite link.
Args:
room_key: The room key or invite link
Returns:
bool: True if successful, False otherwise
"""
# Extract the room key if it's an invite link
if room_key.startswith("pear://keet/"):
room_key = extract_room_key_from_invite(room_key)
return await self.client.join_room(room_key)
async def get_messages(self, room_key: str) -> List[Dict[str, Any]]:
"""
Get messages from a room.
Args:
room_key: The room key or invite link
Returns:
List[Dict[str, Any]]: List of messages
"""
# Extract the room key if it's an invite link
if room_key.startswith("pear://keet/"):
room_key = extract_room_key_from_invite(room_key)
return await self.client.get_messages(room_key)
async def send_message(self, room_key: str, message: str) -> bool:
"""
Send a message to a room.
Args:
room_key: The room key or invite link
message: The message to send
Returns:
bool: True if successful, False otherwise
"""
# Extract the room key if it's an invite link
if room_key.startswith("pear://keet/"):
room_key = extract_room_key_from_invite(room_key)
return await self.client.send_message(room_key, message)
# ToolPlugin interface methods
def get_name(self) -> str:
"""
Get the name of the plugin.
Returns:
str: The plugin name
"""
return "keet"
def get_description(self) -> str:
"""
Get the description of the plugin.
Returns:
str: The plugin description
"""
return "Tools for interacting with the Keet messaging platform"
def get_tools(self) -> Dict[str, Any]:
"""
Get the tools provided by this plugin.
Returns:
Dict[str, Any]: A dictionary of tool names to tool functions
"""
return {
"get_identity": self.get_identity,
"create_identity": self.create_identity,
"get_recent_rooms": self.get_recent_rooms,
"join_room": self.join_room,
"get_messages": self.get_messages,
"send_message": self.send_message
}