LangBot API
Use self.plugin for models, tools, knowledge bases, and storage. For the current event, replies, and conversation state, see Context APIs.
See LLM request configuration for retries, timeouts, numeric reasoning budgets, prompt caching, headers and stream cancellation.
LangBot API
These APIs can be called in any plugin component. Access methods:
- In the plugin root directory
main.py: Internal methods of theselfobject, these APIs are all provided by the plugin class parent classBasePlugin. - In regular plugin component classes: Internal methods of the
self.pluginobject.
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] = {},
timeout: float | None = None,
*,
reasoning_level: ReasoningLevel | None = None,
) -> 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={},
)Reasoning level
All model invocation methods above accept the optional keyword-only reasoning_level parameter:
response = await self.plugin.invoke_llm(
llm_model_uuid=model_uuid,
messages=[provider_message.Message(role="user", content="Analyze this problem")],
reasoning_level="medium",
)Values are provider_default, disabled, enabled, minimal, low, medium, high, xhigh, and max. Available levels depend on the model. LangBot validates support and translates the level into provider parameters; unsupported levels raise an error.
- Omitted or
None: preserves existing model configuration andextra_argsbehavior. provider_default: adds no explicit reasoning level.- Other levels: apply to this call only. Do not also supply provider-specific reasoning parameters through
extra_args.
Runners read their own configuration and explicitly pass the appropriate level for each call, including fallback models and tool follow-ups. LangBot does not extract reasoning levels from Runner configuration. Existing plugins require no changes; when an explicit option is used with an unsupported Host, the SDK requests an upgrade before invoking the model.
Model streaming and usage
| Method | Returns |
|---|---|
invoke_llm_with_usage(llm_model_uuid, messages, funcs=[], extra_args={}, timeout=None, *, reasoning_level=None) | LLMInvokeResult |
invoke_llm_stream(llm_model_uuid, messages, funcs=[], extra_args={}, *, reasoning_level=None) | Async iterator of MessageChunk |
invoke_llm_stream_events(llm_model_uuid, messages, funcs=[], extra_args={}, *, reasoning_level=None) | Async iterator of LLMStreamEvent |
Parameters match invoke_llm. timeout is in seconds; non-streaming calls default to 120 seconds. Use invoke_llm_with_usage when token usage is needed:
from langbot_plugin.api.entities.builtin.provider.message import Message
result = await self.plugin.invoke_llm_with_usage(
llm_model_uuid=model_uuid,
messages=[Message(role="user", content="Hello!")],
)
print(result.message.content)
if result.usage is not None:
print(result.usage.total_tokens)usage may be None; prompt_tokens, completion_tokens, and total_tokens may also be absent. Missing values do not mean zero usage. Additional provider fields, such as cache usage, are preserved.
async for event in self.plugin.invoke_llm_stream_events(
llm_model_uuid=model_uuid,
messages=[Message(role="user", content="Hello!")],
):
if event.chunk is not None:
print(event.chunk.content)
if event.usage is not None:
print(event.usage.model_dump())LLMStreamEvent has optional chunk and usage; the final event may contain usage without text. invoke_llm_stream yields only message/tool-call chunks. MessageChunk.content is the current chunk; all_content, when supplied, is accumulated text. Replace with accumulated text and append only deltas; an is_final chunk may still carry text. Model streaming does not send platform messages; see Run Context APIs for explicit streaming replies.
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_typesList 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)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 regular components, 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)These APIs do not require a query_id and can be used in regular components (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
KnowledgeEnginecomponent classes: Internal methods of theself.pluginobject.
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", ...}}, ...]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"],
)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)Additional resource APIs
These methods are also called through self.plugin.
Invoke a rerank model
async def invoke_rerank(
self, rerank_model_uuid: str, query: str, documents: list[str],
top_k: int | None = None, extra_args: dict[str, Any] | None = None,
timeout: float = 60.0,
) -> list[dict[str, Any]]:
...
scores = await self.plugin.invoke_rerank(
rerank_model_uuid=rerank_model_uuid,
query="How do I configure a bot?",
documents=["Bot configuration guide", "Plugin publishing guide"],
top_k=1,
)Results usually contain the original document index and relevance_score.
Invoke a parser
async def invoke_parser(
self, plugin_author: str, plugin_name: str, storage_path: str,
mime_type: str, filename: str, metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
...
parsers = await self.plugin.list_parsers(mime_type="application/pdf")
if parsers:
parser = parsers[0]
parsed = await self.plugin.invoke_parser(
plugin_author=parser["plugin_author"],
plugin_name=parser["plugin_name"],
storage_path=storage_path,
mime_type="application/pdf",
filename="guide.pdf",
)storage_path is a Host storage path, such as context.file_object.storage_path during knowledge ingestion, not a plugin-local path. The result dictionary contains text, sections, and metadata.
List vector records
async def vector_list(
self, collection_id: str, filters: dict[str, Any] | None = None,
limit: int = 20, offset: int = 0,
) -> dict[str, Any]:
...
page = await self.plugin.vector_list(
collection_id=collection_id,
filters={"file_id": "doc_001"},
limit=20,
offset=0,
)
for item in page["items"]:
print(item["id"], item.get("metadata"))Returns items and total. Items contain id, document, and metadata; total is a best-effort matching count from the storage backend. Filters use the vector API format above.
List plugins and commands
manifests = await self.plugin.list_plugins_manifest()
commands = await self.plugin.list_commands()Both methods take no arguments and return the Host-provided plugin manifest list and command list respectively.
Box sandbox
These methods use self.plugin and the current Workspace. Runner calls automatically retain invocation authorization.
| Method | Returns | Behavior |
|---|---|---|
await self.plugin.get_box_status() | BoxStatus | enabled, available, capacity limit, used, remaining, optional managed required_reuse_key, and unavailable reason |
await self.plugin.list_boxes() | list[BoxSession] | Box id and status snapshot (idle, running, closing) |
await self.plugin.acquire_box(reuse_key, options=None) | BoxSession | Reuse the same key or create within capacity; options supports image |
Unknown capacity fields are None. Zero remaining capacity still permits reuse; Runtime checks creation capacity atomically. The Runner computes reuse keys. When required_reuse_key is supplied, use it. Equal keys in different Workspaces never share a Box.
Use the Runner context to bind the Box before importing attachments or executing tools.