API Reference

This section provides a detailed API reference for the core components of manolo-bot.

AI Components

class manolo_bot.ai.llmbot.LLMBot(llm, bot_config, system_instructions, messages_storage, tools=None)[source]

Bases: object

Base class for a Telegram LLM Chat Bot.

Handles interaction with the LLM, message processing, and context management.

async answer_image_message(chat_id, text, image)[source]

Answer an image message. :type chat_id: int :param chat_id: Chat ID :type text: str :param text: Text to answer :type image: str :param image: Image to answer

Return type:: BaseMessage
Returns:: Response

async answer_message(chat_id, message)[source]

Processes a text message and returns the LLM’s response.

Parameters:

chat_id (int) – The ID of the chat.
message (str) – The text of the message to process.

Return type:

BaseMessage

Returns:

The response message from the LLM.

async answer_voice_message(chat_id, text, audio)[source]: Answer a voice message. :type chat_id: int :param chat_id: Chat ID :type audio: str :param audio: Voice message audio

async answer_webcontent(message_text, response_content)[source]

Answer a web content message. :type message_text: str :param message_text: Text to answer :type response_content: str :param response_content: Response content :param chat_id: Chat ID

Return type:: str | None
Returns:: New response content if the call was successful, None otherwise

bind_tools_on_init = True

async call_sdapi(prompt)[source]

Call the StableDiffusion API. :type prompt: str :param prompt: The prompt to send to the StableDiffusion API.

Return type:: dict | None
Returns:: The response from the StableDiffusion API.

async clean_context()[source]

Clean the chat context.

Return type:: None

async close()[source]

Close all async resources.

Return type:: None

count_tokens(messages)[source]

Count the number of tokens in the messages. :type messages: list[BaseMessage] :param messages: List of messages

Return type:: int
Returns:: Number of tokens

async generate_feedback_message(prompt, max_length=200, chat_id=None)[source]

Generate a feedback message using the LLM.

Parameters:

prompt (str) – Prompt to generate the feedback message
max_length (int) – Maximum length of the feedback message
chat_id (int | None) – Optional chat ID for metadata

Return type:

str

Returns:

Generated feedback message

async generate_image(prompt)[source]

Generate an image. :type prompt: str :param prompt: Prompt to generate the image

Return type:: str | None
Returns:: Image representation in base64 format if the call was successful, None otherwise

async initialize_async_resources()[source]

Initialize all async resources (MCP, etc.).

Return type:: None

async postprocess_response(response, message_text, chat_id)[source]

Postprocess the response from the LLM. :type response: BaseMessage :param response: Response from the LLM :type message_text: str :param message_text: Text of the user message :type chat_id: int :param chat_id: Chat ID return: Final response data

Return type:: dict | None

truncate_chat_context()[source]

Truncate the chat context if it is too long.

Return type:: None

class manolo_bot.ai.llmbot.LLMBuilder(llm_config)[source]

Bases: object

Factory class for creating LangChain Chat Model instances.

get_llm()[source]

Creates and returns an instance of the configured LLM.

Return type:: BaseChatModel
Returns:: A LangChain BaseChatModel instance.
Raises:: Exception – If no LLM configuration is found.

class manolo_bot.ai.llmagent.LLMAgent(llm, bot_config, system_instructions, messages_storage, tools=None)[source]

Bases: LLMBot

Advanced Telegram LLM Chat Bot using a LangGraph-based agent.

This bot can use tools and dynamically integrate with MCP servers.

async answer_image_message(chat_id, text, image)[source]

Answer an image message. :type chat_id: int :param chat_id: Chat ID :type text: str :param text: Text to answer :type image: str :param image: Image to answer

Return type:: BaseMessage
Returns:: Response

async answer_message(chat_id, message)[source]

Processes a text message and returns the LLM’s response.

Parameters:

chat_id (int) – The ID of the chat.
message (str) – The text of the message to process.

Return type:

BaseMessage

Returns:

The response message from the LLM.

async answer_voice_message(chat_id, text, audio)[source]

Answer a voice message. :type chat_id: int :param chat_id: Chat ID :type text: str :param text: Text to answer :type audio: str :param audio: Audio to answer

Return type:: BaseMessage
Returns:: Response

bind_tools_on_init = False

async initialize_async_resources()[source]

Initialize async resources and create agent with all tools.

Return type:: None

Storage Components

class manolo_bot.storage.base.BaseDBHelper[source]

Bases: ABC

async connect()[source]

Connects to the database.

Return type:: None

abstractmethod async disconnect()[source]

Disconnects from the database.

Return type:: None

class manolo_bot.storage.base.BaseMessagesStorage(bot_uuid, chat_id)[source]

Bases: ABC

Abstract base class for message storage.

Provides the interface for persisting and retrieving chat messages.

add_message(message)[source]

Adds a new message.

Return type:: None

abstractmethod async clear_messages()[source]

Clears all messages from the storage.

Return type:: None

abstractmethod async commit()[source]

Include new messages and remove deleted messages from the database asynchronously.

Return type:: None

delete_message(index)[source]

Deletes a message from the storage by index.

Return type:: None

property messages: list[BaseMessage]: Returns a list of non-deleted messages.

abstractmethod async refresh_messages()[source]

Updates the messages list from the database asynchronously.

Return type:: None

class manolo_bot.storage.base.StorageMessage(message, deleted=False, new=False)[source]

Bases: object

deleted: bool = False

message: BaseMessage

new: bool = False

manolo_bot.storage.base.convert_json_to_message(json_message)[source]

Converts a JSON string representation of a message into a BaseMessage instance.

Return type:: BaseMessage

manolo_bot.storage.base.get_messages_key(bot_uuid, chat_id)[source]

Generates a key for storing messages in a database based on bot UUID and chat ID.

Return type:: str

class manolo_bot.storage.memory_storage.MemoryMessagesStorage(bot_uuid, chat_id)[source]

Bases: BaseMessagesStorage

async clear_messages()[source]

Clears all messages from the storage.

Return type:: None

async commit()[source]

Include new messages and remove deleted messages from the database.

Return type:: None

async refresh_messages()[source]

Updates the messages list from the database.

Return type:: None

Configuration

class manolo_bot.ai.config.BotConfig(bot_uuid, bot_name, bot_username, bot_token, user_id, agent_instructions=None, allowed_chat_ids=<factory>, bot_instructions='', bot_instructions_character='', bot_instructions_extra='', simulate_typing=True, simulate_typing_wpm=100, simulate_typing_max_time=10, use_tools=False, enable_mcp=False, mcp_servers_config=<factory>, context_max_tokens=4096, preferred_language='English', add_no_answer=False, is_image_multimodal=False, is_audio_multimodal=False, is_group_assistant=False, agent_mode=False, web_content_request_timeout=10, can_use_tavily_search=False, sdapi_url='', sdapi_params=<factory>, sdapi_negative_prompt='')[source]

Bases: object

add_no_answer: bool = False

agent_instructions: str | None = None

agent_mode: bool = False

allowed_chat_ids: list

bot_instructions: str = ''

bot_instructions_character: str = ''

bot_instructions_extra: str = ''

bot_name: str

bot_token: str

bot_username: str

bot_uuid: str

can_use_tavily_search: bool = False

context_max_tokens: int = 4096

enable_mcp: bool = False

is_audio_multimodal: bool = False

is_group_assistant: bool = False

is_image_multimodal: bool = False

mcp_servers_config: dict

preferred_language: str = 'English'

sdapi_negative_prompt: str = ''

sdapi_params: dict

sdapi_url: str = ''

simulate_typing: bool = True

simulate_typing_max_time: int = 10

simulate_typing_wpm: int = 100

use_tools: bool = False

user_id: int

web_content_request_timeout: int = 10

class manolo_bot.ai.config.LLMConfig(google_api_key, google_api_model, openai_api_key, openai_api_model, openai_api_base_url, ollama_model, rate_limiter_requests_per_second=0.25, rate_limiter_check_every_n_seconds=0.1, rate_limiter_max_bucket_size=10)[source]

Bases: object

google_api_key: str

google_api_model: str

ollama_model: str

openai_api_base_url: str

openai_api_key: str

openai_api_model: str

rate_limiter_check_every_n_seconds: float = 0.1

rate_limiter_max_bucket_size: int = 10

rate_limiter_requests_per_second: float = 0.25

class manolo_bot.config.Config(lazy=False)[source]

Bases: EnvModel

add_no_answer = <envmodel.fields.BooleanField object>

agent_instructions = <envmodel.fields.StringField object>

agent_mode = <envmodel.fields.BooleanField object>

allow_private_chats = <envmodel.fields.BooleanField object>

allowed_chat_ids = <envmodel.fields.StringListField object>

bot_instructions = <envmodel.fields.StringField object>

bot_instructions_character = <envmodel.fields.StringField object>

bot_instructions_extra = <envmodel.fields.StringField object>

bot_name = <envmodel.fields.StringField object>

bot_token = <envmodel.fields.StringField object>

bot_username = <envmodel.fields.StringField object>

bot_uuid = <envmodel.fields.StringField object>

context_max_tokens = <envmodel.fields.IntegerField object>

default_sdapi_params = {'cfg_scale': 1, 'height': 512, 'steps': 1, 'timestep_spacing': 'trailing', 'width': 512}

enable_mcp = <envmodel.fields.BooleanField object>

google_api_key = <envmodel.fields.StringField object>

google_api_model = <envmodel.fields.StringField object>

is_audio_multimodal = <envmodel.fields.BooleanField object>

is_group_assistant = <envmodel.fields.BooleanField object>

is_image_multimodal = <envmodel.fields.BooleanField object>

logging_level = <envmodel.fields.StringField object>

mcp_servers_config = <envmodel.fields.JsonField object>

ollama_model = <envmodel.fields.StringField object>

openai_api_base_url = <envmodel.fields.StringField object>

openai_api_key = <envmodel.fields.StringField object>

openai_api_model = <envmodel.fields.StringField object>

preferred_language = <envmodel.fields.StringField object>

rate_limiter_check_every_n_seconds = <envmodel.fields.FloatField object>

rate_limiter_max_bucket_size = <envmodel.fields.IntegerField object>

rate_limiter_requests_per_second = <envmodel.fields.FloatField object>

redis_url = <envmodel.fields.StringField object>

sdapi_negative_prompt = <envmodel.fields.StringField object>

sdapi_params = <envmodel.fields.JsonField object>

sdapi_url = <envmodel.fields.StringField object>

simulate_typing = <envmodel.fields.BooleanField object>

simulate_typing_max_time = <envmodel.fields.IntegerField object>

simulate_typing_wpm = <envmodel.fields.IntegerField object>

storage_type = <envmodel.fields.StringField object>

use_tavily_search = <envmodel.fields.BooleanField object>

use_tools = <envmodel.fields.BooleanField object>

user_id = <envmodel.fields.IntegerField object>

web_content_request_timeout = <envmodel.fields.IntegerField object>