Skip to content

en plugin dev apis common

Jump to bottom
langbot-wiki-sync[bot] edited this page Jul 29, 2026 · 1 revision

Plugin Common APIs

LangBot provides a series of APIs for plugins to operate various LangBot modules and control message contexts.

Request API

Operations for the currently processing user request (message). Only available in EventListener and Command components. Access methods:

  • In event handler methods of EventListener: Internal methods of event_context: context.EventContext object
  • In subcommand handler methods of Command: Internal methods of context: ExecuteContext object

Direct Reply Message

Directly reply with a message chain to the session where the current request is located.

For message chain construction methods, please refer to Message Platform Entities.

async def reply(
    self, message_chain: platform_message.MessageChain, quote_origin: bool = False
):
    """Reply to the message request

    Args:
        message_chain (platform.types.MessageChain): LangBot message chain
        quote_origin (bool): Whether to quote the original message
    """

# Usage example
await event_context.reply(
    platform_message.MessageChain([
        platform_message.Plain(text="Hello, world!"),
    ]),
)

Get Bot UUID

Get the UUID of the bot that originated the current request.

async def get_bot_uuid(self) -> str:
    """Get the bot uuid"""

# Usage example
bot_uuid = await event_context.get_bot_uuid()

Set Request Variables

Some information in a single request is stored in request variables. When using external Agent platforms like Dify, these variables are explicitly passed to Agent platforms.

async def set_query_var(self, key: str, value: Any):
    """Set a query variable"""

# Usage example
await event_context.set_query_var("key", "value")

Get Request Variables

Get a single request variable.

async def get_query_var(self, key: str) -> Any:
    """Get a query variable"""

# Usage example
query_var = await event_context.get_query_var("key")

Get All Request Variables

async def get_query_vars(self) -> dict[str, Any]:
    """Get all query variables"""

# Usage example
query_vars = await event_context.get_query_vars()

List Pipeline Knowledge Bases

Get the list of knowledge bases configured for the pipeline used by the current request.

async def list_pipeline_knowledge_bases(self) -> list[dict[str, Any]]:
    """List knowledge bases configured for the current pipeline"""

# Usage example
knowledge_bases = await event_context.list_pipeline_knowledge_bases()

# Return example
[
    {
        "uuid": "kb_uuid",
        "name": "Product Docs",
        "description": "Internal product documentation",
    }
]

Only returns knowledge bases bound in the Local Agent configuration of the pipeline corresponding to the current request. If the current pipeline has no knowledge bases configured, or does not use Local Agent, an empty list is returned.

LangBot API

These APIs can be called in any plugin component. Access methods:

  • In the plugin root directory main.py: Internal methods of the self object, these APIs are all provided by the plugin class parent class BasePlugin.
  • In any plugin component class: Internal methods of the self.plugin object.

Get Plugin Configuration

Plugin configuration format can be written in manifest.yaml, and users need to fill it out according to the plugin configuration format in LangBot's plugin management. Plugin code can then call this API to get plugin configuration information.

def get_config(self) -> dict[str, typing.Any]:
    """Get the config of the plugin."""

# Usage example
config = self.plugin.get_config()

Get LangBot Version

Get the LangBot version number, returned as a string in format v<major>.<minor>.<patch>.

async def get_langbot_version(self) -> str:
    """Get the langbot version"""

# Usage example
langbot_version = await self.plugin.get_langbot_version()

Get Configured Bot List

Returns a list of all bot UUIDs.

async def get_bots(self) -> list[str]:
    """Get all bots"""

# Usage example
bots = await self.plugin.get_bots()

Get Bot Information

Get bot information.

async def get_bot_info(self, bot_uuid: str) -> dict[str, Any]:
    """Get a bot info"""

# Usage example
bot_info = await self.plugin.get_bot_info("de639861-be05-4018-859b-c2e2d3e0d603")

# Return example
{
    "uuid": "de639861-be05-4018-859b-c2e2d3e0d603",
    "name": "aiocqhttp",
    "description": "Migrated from LangBot v3",
    "adapter": "aiocqhttp",
    "enable": true,
    "use_pipeline_name": "ChatPipeline",
    "use_pipeline_uuid": "c30a1dca-e91c-452b-83ec-84d635a30028",
    "created_at": "2025-05-10T13:53:08",
    "updated_at": "2025-08-12T11:27:30",
    "adapter_runtime_values": {  # Present if the bot is currently running
        "bot_account_id": 960164003  # Bot account ID
    }
}

Send Proactive Message

Send proactive messages through bot UUID and target session ID.

For message chain construction methods, please refer to Message Platform Entities.

async def send_message(
    self,
    bot_uuid: str,
    target_type: str,
    target_id: str,
    message_chain: platform_message.MessageChain,
) -> None:
    """Send a message to a session"""

# Usage example
await self.plugin.send_message(
    bot_uuid="de639861-be05-4018-859b-c2e2d3e0d603",
    target_type="person",
    target_id="1010553892",
    message_chain=platform_message.MessageChain([platform_message.Plain(text="Hello, world!")]),
)

Get Configured LLM Model List

Returns a list of UUIDs for all configured LLM models.

async def get_llm_models(self) -> list[str]:
    """Get all LLM models"""

# Usage example
llm_models = await self.plugin.get_llm_models()

Invoke LLM Model

Invoke an LLM model, returns an LLM message. Non-streaming.

async def invoke_llm(
    self,
    llm_model_uuid: str,
    messages: list[provider_message.Message],
    funcs: list[resource_tool.LLMTool] = [],
    extra_args: dict[str, Any] = {},
) -> provider_message.Message:
    """Invoke an LLM model"""

# Usage example
llm_message = await self.plugin.invoke_llm(
    llm_model_uuid="llm_model_uuid",
    messages=[provider_message.Message(role="user", content="Hello, world!")],
    funcs=[],
    extra_args={},
)

List Available Parsers

List Parser plugins currently available on the host, optionally filtered by MIME type.

async def list_parsers(self, mime_type: str | None = None) -> list[dict[str, Any]]:
    """List available Parser plugins"""

# Usage example
parsers = await self.plugin.list_parsers(mime_type="application/pdf")
# Each item includes plugin_id, plugin_author, plugin_name, name, description, supported_mime_types

List All Available Tools

List all available tools in the current LangBot instance (including plugin tools and MCP tools).

async def list_tools(self) -> list[dict[str, Any]]:
    """List all available tools

    Returns:
        A list of tool dicts, each containing:
        - name: Tool name
        - label: Display label (i18n)
        - description: Tool description (i18n)
        - icon: Tool icon
        - spec: Tool specification (includes llm_prompt and parameters)
    """

# Example
tools = await self.plugin.list_tools()
for tool in tools:
    print(f"Tool: {tool['name']}")

Get Tool Detail

Get detailed information about a specific tool.

async def get_tool_detail(self, tool_name: str) -> dict[str, Any]:
    """Get detailed information about a specific tool

    Args:
        tool_name: Tool name (metadata.name, e.g. "get_weather_alerts")

    Returns:
        Tool detail dict containing name, label, description, spec (with parameters and llm_prompt)
    """

# Example
detail = await self.plugin.get_tool_detail("get_weather_alerts")
print(detail['spec']['parameters'])

Call Tool

Call a specific tool.

async def call_tool(
    self,
    tool_name: str,
    parameters: dict[str, Any],
    session: dict[str, Any],
    query_id: int,
) -> dict[str, Any]:
    """Call a specific tool

    Args:
        tool_name: Tool name (metadata.name)
        parameters: Tool parameters
        session: Session info
        query_id: Query ID

    Returns:
        Tool response dict
    """

# Example
result = await self.plugin.call_tool(
    tool_name="get_weather_alerts",
    parameters={"state": "CA"},
    session={},
    query_id=0,
)
print(result)

Tip

Tool names use metadata.name (e.g. echo_tool, get_weather_alerts), without author or plugin name prefix.

Set Plugin Persistent Data

Persistently store plugin data. Data stored through this interface can only be accessed by this plugin. Values need to be converted to bytes manually.

async def set_plugin_storage(self, key: str, value: bytes) -> None:
    """Set a plugin storage value"""

# Usage example
await self.plugin.set_plugin_storage("key", b"value")

Get Plugin Persistent Data

async def get_plugin_storage(self, key: str) -> bytes:
    """Get a plugin storage value"""

# Usage example
plugin_storage = await self.plugin.get_plugin_storage("key")

Get All Plugin Persistent Data Keys

async def get_plugin_storage_keys(self) -> list[str]:
    """Get all plugin storage keys"""

# Usage example
plugin_storage_keys = await self.plugin.get_plugin_storage_keys()

Delete Plugin Persistent Data

async def delete_plugin_storage(self, key: str) -> None:
    """Delete a plugin storage value"""

# Usage example
await self.plugin.delete_plugin_storage("key")

Get Workspace Persistent Data

Data stored through this interface can be accessed by all plugins. Values need to be converted to bytes manually.

async def set_workspace_storage(self, key: str, value: bytes) -> None:
    """Set a workspace storage value"""

# Usage example
await self.plugin.set_workspace_storage("key", b"value")

Get Workspace Persistent Data

async def get_workspace_storage(self, key: str) -> bytes:
    """Get a workspace storage value"""

# Usage example
workspace_storage = await self.plugin.get_workspace_storage("key")

Get All Workspace Persistent Data Keys

async def get_workspace_storage_keys(self) -> list[str]:
    """Get all workspace storage keys"""

# Usage example
workspace_storage_keys = await self.plugin.get_workspace_storage_keys()

Delete Workspace Persistent Data

async def delete_workspace_storage(self, key: str) -> None:
    """Delete a workspace storage value"""

# Usage example
await self.plugin.delete_workspace_storage("key")

Get Plugin File-typed Config Field Data

async def get_config_file(self, file_key: str) -> bytes:
    """Get a config file value"""

# Usage example
file_bytes = await self.plugin.get_config_file("key")

Use this in conjunction with configuration fields of type file or array[file.

Knowledge Base API

These APIs are accessible via self.plugin in any component, allowing you to list and retrieve from all knowledge bases in the LangBot instance without pipeline restrictions.

List All Knowledge Bases

List all available knowledge bases in the LangBot instance.

async def list_knowledge_bases(self) -> list[dict[str, Any]]:
    """List all knowledge bases

    Returns:
        List of knowledge base dicts, each containing:
        - uuid: Knowledge base UUID
        - name: Knowledge base name
        - description: Knowledge base description
    """

# Usage example
knowledge_bases = await self.plugin.list_knowledge_bases()
for kb in knowledge_bases:
    print(f"KB: {kb['name']} ({kb['uuid']})")

Retrieve from Knowledge Base

Retrieve relevant documents from any knowledge base.

async def retrieve_knowledge(
    self,
    kb_id: str,
    query_text: str,
    top_k: int = 5,
    filters: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    """Retrieve from a knowledge base

    Args:
        kb_id: Knowledge base UUID (from list_knowledge_bases)
        query_text: Search query text
        top_k: Number of results to return (default: 5)
        filters: Optional metadata filters for retrieval

    Returns:
        List of retrieval result entries
    """

# Usage example
results = await self.plugin.retrieve_knowledge(
    kb_id="kb-uuid-here",
    query_text="How to configure the system?",
    top_k=3,
)
for entry in results:
    print(entry)

Tip

These APIs do not require a query_id and can be used in any component (including Tool components). They can access all knowledge bases without being restricted to the current pipeline's configuration.

RAG API

These APIs are available for KnowledgeEngine components to access the LangBot host's embedding models, vector database, and file storage. Access method:

  • In KnowledgeEngine component classes: Internal methods of the self.plugin object.

Invoke Embedding Model

Generate text embeddings using the host's configured embedding model.

async def invoke_embedding(
    self,
    embedding_model_uuid: str,
    texts: list[str],
) -> list[list[float]]:
    """Generate embeddings using host's embedding model

    Args:
        embedding_model_uuid: Embedding model UUID
        texts: List of texts to embed

    Returns:
        List of embedding vectors, one per input text
    """

# Usage example
vectors = await self.plugin.invoke_embedding("model_uuid", ["Hello", "World"])

Vector Upsert

Upsert vectors to the host's vector database.

async def vector_upsert(
    self,
    collection_id: str,
    vectors: list[list[float]],
    ids: list[str],
    metadata: list[dict[str, Any]] | None = None,
    documents: list[str] | None = None,
) -> None:
    """Upsert vectors

    Args:
        collection_id: Target collection ID
        vectors: List of vectors
        ids: List of unique IDs for vectors
        metadata: Optional list of metadata dicts
        documents: Optional raw text documents. Required for full-text
            and hybrid search in backends that support them.
    """

# Usage example
await self.plugin.vector_upsert(
    collection_id="kb_uuid",
    vectors=[[0.1, 0.2, ...], [0.3, 0.4, ...]],
    ids=["chunk_0", "chunk_1"],
    metadata=[{"document_id": "doc1"}, {"document_id": "doc1"}],
    documents=["chunk text 0", "chunk text 1"],
)

Vector Search

Search similar vectors in the host's vector database.

async def vector_search(
    self,
    collection_id: str,
    query_vector: list[float],
    top_k: int = 5,
    filters: dict[str, Any] | None = None,
    search_type: str = "vector",
    query_text: str = "",
) -> list[dict[str, Any]]:
    """Vector search

    Args:
        collection_id: Target collection ID
        query_vector: Query vector for similarity search
        top_k: Number of results to return
        filters: Optional metadata filters
        search_type: One of 'vector', 'full_text', 'hybrid'
        query_text: Raw query text, used for full_text and hybrid search

    Returns:
        List of search results (dict with id, score, metadata, etc.)
    """

# Usage example
results = await self.plugin.vector_search(
    collection_id="kb_uuid",
    query_vector=[0.1, 0.2, ...],
    top_k=5,
    search_type="hybrid",
    query_text="search query",
)
# Return format: [{"id": "chunk_0", "score": 0.123, "metadata": {"document_id": "doc1", ...}}, ...]

Note

Each result returned by vector_search is a dict containing id (vector ID), score (distance score), and metadata (metadata provided during upsert). If you need text content in retrieval results, store the text in metadata during ingestion.

Vector Delete

Delete vectors from the host's vector database.

async def vector_delete(
    self,
    collection_id: str,
    file_ids: list[str] | None = None,
    filters: dict[str, Any] | None = None,
) -> int:
    """Vector delete

    Args:
        collection_id: Target collection ID
        file_ids: File IDs whose vectors should be deleted
        filters: Optional metadata filters for deletion

    Returns:
        Number of deleted items
    """

# Usage example
deleted = await self.plugin.vector_delete(
    collection_id="kb_uuid",
    file_ids=["doc_001"],
)

Note

The filters parameter supports Chroma-style where syntax for metadata filtering. Multiple top-level keys are AND-ed. Supported operators: $eq, $ne, $gt, $gte, $lt, $lte, $in, $nin.

# Implicit $eq
results = await self.plugin.vector_search(
    collection_id="kb_uuid",
    query_vector=[0.1, 0.2, ...],
    filters={"file_id": "abc"},
)

# Comparison operator
results = await self.plugin.vector_search(
    collection_id="kb_uuid",
    query_vector=[0.1, 0.2, ...],
    filters={"created_at": {"$gte": 1700000000}},
)

# In-list operator
results = await self.plugin.vector_search(
    collection_id="kb_uuid",
    query_vector=[0.1, 0.2, ...],
    filters={"file_type": {"$in": ["pdf", "docx"]}},
)

# Delete by filter
deleted = await self.plugin.vector_delete(
    collection_id="kb_uuid",
    filters={"file_type": {"$eq": "pdf"}},
)

Note: Chroma, Qdrant, and SeekDB store full metadata and can filter on any field. Milvus and pgvector only store text, file_id, and chunk_uuid — filters on other fields will be silently ignored.

Get uploaded file content from the host's storage.

async def get_knowledge_file_stream(self, storage_path: str) -> bytes:
    """Get file content

    Args:
        storage_path: File storage path (from FileObject.storage_path)

    Returns:
        File content as bytes
    """

# Usage example
file_bytes = await self.plugin.get_knowledge_file_stream(context.file_object.storage_path)

Automatically synchronized from langbot-app/langbot-wiki.

LangBot Documentation

Home

简体中文
指南
开发者
API 参考
Other pages
English
Guides
Developers
API Reference
Other pages
日本語
ガイド
開発者
API リファレンス
Other pages

Clone this wiki locally