API Reference

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

AI Components

exception manolo_bot.ai.llmbot.FileTooLargeError[source]

Bases: ValueError

Exception raised when a file exceeds the allowed size.

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

Bases: object

Base class for a Telegram LLM Chat Bot.

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

async answer_document_message(chat_id, text, document_url, filename)[source]

Answer a document message.

Parameters:

chat_id (int) – Chat ID
text (str) – Text to answer (user prompt/caption)
document_url (str) – URL to download the document
filename (str) – Original filename

Return type:

BaseMessage

Returns:

Response

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

property system_instructions: list[langchain_core.messages.BaseMessage]: Get the system instructions mapping.

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, document_storage=None, system_instructions_mapping=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_document_message(chat_id, text, document_url, filename)[source]

Answer a document message using the agent.

Parameters:

chat_id (int) – Chat ID
text (str) – Text to answer
document_url (str) – Document URL
filename (str) – Original filename

Return type:

BaseMessage

Returns:

Response

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

MCP Manager for handling Model Context Protocol servers.

class manolo_bot.ai.mcp_manager.MCPManager(config)[source]

Bases: object

Manages MCP server connections and tool loading.

async connect()[source]

Initialize MCP client and connect to configured servers.

Return type:: None

async disconnect()[source]

Close all MCP server connections.

Return type:: None

async get_tools()[source]

Get all loaded MCP tools.

Return type:: list[BaseTool]

property is_connected: bool: Check if MCP is connected.

class manolo_bot.ai.document_loaders.DocumentLoader[source]

Bases: object

Utility class for extracting text from different document formats using LangChain native parsers.

SUPPORTED_EXTENSIONS = ['pdf', 'docx', 'txt', 'md', 'csv']

extract_text(file_content, filename)[source]

Dispatcher method to extract text based on file extension.

Return type:: str

extract_text_from_docx(file)[source]

Extracts text from a DOCX file using DocxParser.

Return type:: str

extract_text_from_pdf(file)[source]

Extracts text from a PDF file using PyPDFParser.

Return type:: str

extract_text_from_txt(file)[source]

Extracts text from a TXT/MD/CSV file using TextParser.

Return type:: str

classmethod validate_filename(filename)[source]

Validates if a filename has a supported extension.

Parameters:: filename (str) – The filename to validate.
Raises:: UnsupportedFileError – If the extension is not supported.
Return type:: None

class manolo_bot.ai.document_loaders.DocxParser(*args, **kwargs)[source]

Bases: BaseBlobParser

Parser for DOCX files using python-docx. Follows the LangChain BaseBlobParser interface. We use this instead of MsWordParser to avoid the heavy ‘unstructured’ dependency.

lazy_parse(blob)[source]

Return type:: Iterator[Document]

exception manolo_bot.ai.document_loaders.UnsupportedFileError[source]

Bases: ValueError

Exception raised when a file format is not supported.

manolo_bot.ai.document_loaders.clean_text(text)[source]

Basic text cleaning to reduce token usage. Removes multiple whitespaces and newlines.

Return type:: str

Storage Components

class manolo_bot.storage.messages.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.messages.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[langchain_core.messages.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.messages.base.StorageMessage(message, deleted=False, new=False)[source]

Bases: object

deleted: bool = False

message: langchain_core.messages.BaseMessage

new: bool = False

manolo_bot.storage.messages.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.messages.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.messages.memory.MemoryMessagesStorage(bot_uuid, chat_id)[source]

Bases: BaseMessagesStorage

In-memory implementation of message storage.

async clear_messages()[source]

Clears all messages from the memory storage for the current chat.

Return type:: None

async commit()[source]

Include new messages and remove deleted messages from the memory storage.

Return type:: None

async refresh_messages()[source]

Updates the messages list from the memory storage.

Return type:: None

class manolo_bot.storage.messages.redis.RedisDBHelper(db_url)[source]

Bases: BaseDBHelper

Helper class for Redis database operations.

async connect()[source]

Connects to the Redis database.

Return type:: None

async disconnect()[source]

Disconnects from the Redis database.

Return type:: None

class manolo_bot.storage.messages.redis.RedisMessagesStorage(db, bot_uuid, chat_id)[source]

Bases: BaseMessagesStorage

Redis-based implementation of message storage.

async clear_messages()[source]

Clears all messages from the Redis database for the current chat.

Return type:: None

async commit()[source]

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

Return type:: None

async refresh_messages()[source]

Updates the messages list from the Redis database.

Return type:: None

class manolo_bot.storage.documents.base.BaseDocumentStorage(bot_uuid)[source]

Bases: ABC

Abstract base class for document storage.

Provides the interface for persisting and retrieving extracted document text.

abstractmethod async clear(chat_id)[source]

Clears all stored documents for a specific chat.

Parameters:: chat_id (int) – The ID of the chat.
Return type:: None

abstractmethod async list_documents(chat_id)[source]

Lists all stored documents for a specific chat.

Parameters:: chat_id (int) – The ID of the chat.
Return type:: list[str]
Returns:: A list of filenames.

abstractmethod async retrieve(chat_id, filename)[source]

Retrieves the extracted text of a document.

Parameters:

chat_id (int) – The ID of the chat.
filename (str) – The name of the document.

Return type:

str | None

Returns:

The extracted text or None if not found.

abstractmethod async store(chat_id, filename, text)[source]

Stores the extracted text of a document.

Parameters:

chat_id (int) – The ID of the chat.
filename (str) – The name of the document.
text (str) – The extracted text.

Return type:

None

class manolo_bot.storage.documents.file.FileDocumentStorage(bot_uuid, base_path=None)[source]

Bases: BaseDocumentStorage

File-based implementation of document storage.

async clear(chat_id)[source]

Clears all stored documents for a specific chat from the filesystem.

Parameters:: chat_id (int) – The ID of the chat.
Return type:: None

async list_documents(chat_id)[source]

Lists all stored documents for a specific chat in the filesystem.

Parameters:: chat_id (int) – The ID of the chat.
Return type:: list[str]
Returns:: A list of filenames.

async retrieve(chat_id, filename)[source]

Retrieves the extracted text of a document from the filesystem.

Parameters:

chat_id (int) – The ID of the chat.
filename (str) – The name of the document.

Return type:

str | None

Returns:

The extracted text or None if not found or path is insecure.

async store(chat_id, filename, text)[source]

Stores the extracted text of a document in the filesystem.

Parameters:

chat_id (int) – The ID of the chat.
filename (str) – The name of the document.
text (str) – The extracted text.

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_document_multimodal=False, is_group_assistant=False, agent_mode=False, web_content_request_timeout=10, max_document_size=2097152, 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_document_multimodal: bool = False

is_group_assistant: bool = False

is_image_multimodal: bool = False

max_document_size: int = 2097152

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}

document_storage_path = <envmodel.fields.StringField object>

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_document_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>

max_document_size = <envmodel.fields.IntegerField 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>