xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-09-01 15:17:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into dev/4.11.x

Browse Source 
# Conflicts:
#	src/langbot/pkg/pipeline/preproc/preproc.py
#	src/langbot/pkg/pipeline/process/handlers/chat.py
#	src/langbot/pkg/provider/runners/localagent.py
#	src/langbot/pkg/provider/tools/toolmgr.py
#	src/langbot/templates/metadata/pipeline/ai.yaml
#	tests/unit_tests/test_preproc.py
#	web/src/app/home/components/dynamic-form/DynamicFormComponent.tsx
#	web/src/app/home/pipelines/components/pipeline-form/PipelineFormComponent.tsx
This commit is contained in:
Junyan Qin
2026-06-30 21:07:13 +08:00
parent db1879164d 7de7c7f714
commit 1c9b54d02c
167 changed files with 16186 additions and 1722 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py
+8
View File
@@ -86,6 +86,10 @@ class PipelinesRouterGroup(group.RouterGroup):
'available_plugins': plugins,
'bound_mcp_servers': extensions_prefs.get('mcp_servers', []),
'available_mcp_servers': mcp_servers,
'bound_mcp_resources': extensions_prefs.get('mcp_resources', []),
'mcp_resource_agent_read_enabled': extensions_prefs.get(
'mcp_resource_agent_read_enabled', True
),
'bound_skills': extensions_prefs.get('skills', []),
'available_skills': available_skills,
}
@@ -99,6 +103,8 @@ class PipelinesRouterGroup(group.RouterGroup):
bound_plugins = json_data.get('bound_plugins', [])
bound_mcp_servers = json_data.get('bound_mcp_servers', [])
bound_skills = json_data.get('bound_skills', [])
bound_mcp_resources = json_data.get('bound_mcp_resources')
mcp_resource_agent_read_enabled = json_data.get('mcp_resource_agent_read_enabled')
await self.ap.pipeline_service.update_pipeline_extensions(
pipeline_uuid,
@@ -108,6 +114,8 @@ class PipelinesRouterGroup(group.RouterGroup):
enable_all_mcp_servers,
bound_skills=bound_skills,
enable_all_skills=enable_all_skills,
bound_mcp_resources=bound_mcp_resources,
mcp_resource_agent_read_enabled=mcp_resource_agent_read_enabled,
)
return self.success()
src/langbot/pkg/api/http/controller/groups/platform/bots.py
+27 -11
View File
@@ -18,7 +18,6 @@ class BotsRouterGroup(group.RouterGroup):
@self.route('/<bot_uuid>', methods=['GET', 'PUT', 'DELETE'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
async def _(bot_uuid: str) -> str:
if quart.request.method == 'GET':
# 返回运行时信息，包括webhook地址等
bot = await self.ap.bot_service.get_runtime_bot_info(bot_uuid)
if bot is None:
return self.http_status(404, -1, 'bot not found')
@@ -37,30 +36,21 @@ class BotsRouterGroup(group.RouterGroup):
from_index = json_data.get('from_index', -1)
max_count = json_data.get('max_count', 10)
logs, total_count = await self.ap.bot_service.list_event_logs(bot_uuid, from_index, max_count)
return self.success(
data={
'logs': logs,
'total_count': total_count,
}
)
return self.success(data={'logs': logs, 'total_count': total_count})
@self.route('/<bot_uuid>/send_message', methods=['POST'], auth_type=group.AuthType.API_KEY)
async def _(bot_uuid: str) -> str:
"""Send message to a specific target via bot"""
json_data = await quart.request.json
target_type = json_data.get('target_type')
target_id = json_data.get('target_id')
message_chain_data = json_data.get('message_chain')
# Validate required fields
if not target_type:
return self.http_status(400, -1, 'target_type is required')
if not target_id:
return self.http_status(400, -1, 'target_id is required')
if not message_chain_data:
return self.http_status(400, -1, 'message_chain is required')
# Validate target_type
if target_type not in ['person', 'group']:
return self.http_status(400, -1, 'target_type must be either "person" or "group"')
@@ -72,3 +62,29 @@ class BotsRouterGroup(group.RouterGroup):
traceback.print_exc()
return self.http_status(500, -1, f'Failed to send message: {str(e)}')
# ============ Bot Admins ============
@self.route('/<bot_uuid>/admins', methods=['GET', 'POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
async def _(bot_uuid: str) -> str:
if quart.request.method == 'GET':
admins = await self.ap.bot_service.get_bot_admins(bot_uuid)
return self.success(data={'admins': admins})
elif quart.request.method == 'POST':
json_data = await quart.request.json
launcher_type = json_data.get('launcher_type', '').strip()
launcher_id = str(json_data.get('launcher_id', '')).strip()
if not launcher_type or not launcher_id:
return self.http_status(400, -1, 'launcher_type and launcher_id are required')
try:
admin_id = await self.ap.bot_service.add_bot_admin(bot_uuid, launcher_type, launcher_id)
return self.success(data={'id': admin_id})
except Exception as e:
return self.http_status(409, -1, str(e))
@self.route(
'/<bot_uuid>/admins/<int:admin_id>', methods=['DELETE'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY
)
async def _(bot_uuid: str, admin_id: int) -> str:
await self.ap.bot_service.delete_bot_admin(bot_uuid, admin_id)
return self.success()
src/langbot/pkg/api/http/controller/groups/resources/mcp.py
+48
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import quart
import traceback
from urllib.parse import unquote
from ... import group
@@ -66,3 +67,50 @@ class MCPRouterGroup(group.RouterGroup):
server_data = await quart.request.json
task_id = await self.ap.mcp_service.test_mcp_server(server_name=server_name, server_data=server_data)
return self.success(data={'task_id': task_id})
@self.route('/servers/<server_name>/resources', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
async def _(server_name: str) -> str:
"""Get resources from an MCP server"""
server_name = unquote(server_name)
try:
resources = await self.ap.mcp_service.get_mcp_server_resources(server_name)
templates = await self.ap.mcp_service.get_mcp_server_resource_templates(server_name)
runtime_info = await self.ap.mcp_service.get_runtime_info(server_name)
return self.success(
data={
'resources': resources,
'resource_templates': templates,
'resource_capabilities': (runtime_info or {}).get('resource_capabilities', {}),
}
)
except Exception as e:
return self.http_status(500, -1, f'Failed to get resources: {str(e)}')
@self.route('/servers/<server_name>/resource-templates', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
async def _(server_name: str) -> str:
"""Get resource templates from an MCP server"""
server_name = unquote(server_name)
try:
templates = await self.ap.mcp_service.get_mcp_server_resource_templates(server_name)
return self.success(data={'resource_templates': templates})
except Exception as e:
return self.http_status(500, -1, f'Failed to get resource templates: {str(e)}')
@self.route('/servers/<server_name>/resources/read', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
async def _(server_name: str) -> str:
"""Read a resource from an MCP server"""
server_name = unquote(server_name)
data = await quart.request.json
uri = data.get('uri')
if not uri:
return self.http_status(400, -1, 'URI is required')
try:
envelope = await self.ap.mcp_service.read_mcp_server_resource_envelope(
server_name,
uri,
max_bytes=data.get('max_bytes'),
include_blob=bool(data.get('include_blob', False)),
)
return self.success(data=envelope)
except Exception as e:
return self.http_status(500, -1, f'Failed to read resource: {str(e)}')
src/langbot/pkg/api/http/controller/groups/resources/tools.py
+31 -13
View File
@@ -1,5 +1,7 @@
from __future__ import annotations
import quart
from ... import group
@@ -9,25 +11,41 @@ class ToolsRouterGroup(group.RouterGroup):
@self.route('', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
async def _() -> str:
"""获取所有可用工具列表"""
tools = await self.ap.tool_mgr.get_all_tools()
pipeline_uuid = quart.request.args.get('pipeline_uuid') or quart.request.args.get('pipeline_id')
bound_plugins: list[str] | None = None
bound_mcp_servers: list[str] | None = None
tool_list = []
for tool in tools:
tool_list.append(
{
'name': tool.name,
'description': tool.description,
'human_desc': tool.human_desc,
'parameters': tool.parameters,
}
)
if pipeline_uuid:
pipeline = await self.ap.pipeline_service.get_pipeline(pipeline_uuid)
if pipeline is None:
return self.http_status(404, -1, 'pipeline not found')
return self.success(data={'tools': tool_list})
extensions_prefs = pipeline.get('extensions_preferences', {}) or {}
if not extensions_prefs.get('enable_all_plugins', True):
bound_plugins = [
f'{plugin.get("author", "")}/{plugin.get("name", "")}'
for plugin in extensions_prefs.get('plugins', [])
if isinstance(plugin, dict) and plugin.get('name')
]
if not extensions_prefs.get('enable_all_mcp_servers', True):
bound_mcp_servers = [
server for server in (extensions_prefs.get('mcp_servers', []) or []) if isinstance(server, str)
]
return self.success(
data={
'tools': await self.ap.tool_mgr.get_tool_catalog(
bound_plugins,
bound_mcp_servers,
include_skill_authoring=True,
)
}
)
@self.route('/<tool_name>', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
async def _(tool_name: str) -> str:
"""获取特定工具详情"""
tools = await self.ap.tool_mgr.get_all_tools()
tools = await self.ap.tool_mgr.get_all_tools(include_skill_authoring=True)
for tool in tools:
if tool.name == tool_name:
src/langbot/pkg/api/http/controller/groups/survey.py
+46
View File
@@ -1,3 +1,5 @@
import base64
import quart
from .. import group
@@ -30,6 +32,50 @@ class SurveyRouterGroup(group.RouterGroup):
return self.fail(2, 'Failed to submit response')
return self.fail(3, 'Survey not available')
@self.route('/feedback', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
async def _feedback(user_email: str) -> str:
"""Submit on-demand user feedback from the sidebar."""
json_data = await quart.request.get_json(silent=True) or {}
content = str(json_data.get('content', '')).strip()
attachments = json_data.get('attachments', [])
if not content:
return self.fail(1, 'content required')
if len(content) > 5000:
return self.fail(2, 'content too long')
if not isinstance(attachments, list):
return self.fail(3, 'attachments must be an array')
if len(attachments) > 3:
return self.fail(4, 'too many attachments')
normalized_attachments = []
for item in attachments:
if not isinstance(item, dict):
continue
data_url = str(item.get('data_url', ''))
mime_type = str(item.get('mime_type', ''))[:128]
name = str(item.get('name', ''))[:255]
if not data_url.startswith('data:image/'):
continue
try:
payload = data_url.split(',', 1)[1]
if len(base64.b64decode(payload, validate=True)) > 1024 * 1024:
return self.fail(5, 'attachment too large')
except Exception:
return self.fail(5, 'attachment too large')
normalized_attachments.append({'name': name, 'mime_type': mime_type, 'data_url': data_url})
if self.ap.survey:
ok = await self.ap.survey.submit_feedback(
content=content,
attachments=normalized_attachments,
user_email=user_email,
)
if ok:
return self.success()
return self.fail(6, 'Failed to submit feedback')
return self.fail(7, 'Survey not available')
@self.route('/dismiss', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
async def _dismiss() -> str:
"""Dismiss survey."""
src/langbot/pkg/api/http/controller/groups/user.py
+7
View File
@@ -195,6 +195,13 @@ class UserRouterGroup(group.RouterGroup):
@self.route('/set-password', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
async def _(user_email: str) -> str:
"""Set password for Space account (first time) or change password"""
# Check if modifying login info is allowed
allow_modify_login_info = self.ap.instance_config.data.get('system', {}).get(
'allow_modify_login_info', True
)
if not allow_modify_login_info:
return self.http_status(403, -1, 'Modifying login info is disabled')
json_data = await quart.request.json
new_password = json_data.get('new_password')
current_password = json_data.get('current_password')
src/langbot/pkg/api/http/service/bot.py
+32
View File
@@ -274,3 +274,35 @@ class BotService:
# Send message via adapter
await runtime_bot.adapter.send_message(target_type, str(target_id), message_chain)
# ============ Bot Admins ============
async def get_bot_admins(self, bot_uuid: str) -> list[dict]:
from ....entity.persistence import bot as persistence_bot
result = await self.ap.persistence_mgr.execute_async(
sqlalchemy.select(persistence_bot.BotAdmin).where(persistence_bot.BotAdmin.bot_uuid == bot_uuid)
)
return [{'id': r.id, 'launcher_type': r.launcher_type, 'launcher_id': r.launcher_id} for r in result.all()]
async def add_bot_admin(self, bot_uuid: str, launcher_type: str, launcher_id: str) -> int:
from ....entity.persistence import bot as persistence_bot
result = await self.ap.persistence_mgr.execute_async(
sqlalchemy.insert(persistence_bot.BotAdmin).values(
bot_uuid=bot_uuid,
launcher_type=launcher_type,
launcher_id=launcher_id,
)
)
return result.inserted_primary_key[0]
async def delete_bot_admin(self, bot_uuid: str, admin_id: int) -> None:
from ....entity.persistence import bot as persistence_bot
await self.ap.persistence_mgr.execute_async(
sqlalchemy.delete(persistence_bot.BotAdmin).where(
persistence_bot.BotAdmin.bot_uuid == bot_uuid,
persistence_bot.BotAdmin.id == admin_id,
)
)
src/langbot/pkg/api/http/service/mcp.py
+26
View File
@@ -136,6 +136,32 @@ class MCPService:
if server_name in self.ap.tool_mgr.mcp_tool_loader.sessions:
await self.ap.tool_mgr.mcp_tool_loader.remove_mcp_server(server_name)
async def get_mcp_server_resources(self, server_name: str) -> list[dict]:
"""Get resources from a specific MCP server."""
return await self.ap.tool_mgr.mcp_tool_loader.get_resources(server_name)
async def get_mcp_server_resource_templates(self, server_name: str) -> list[dict]:
"""Get resource templates from a specific MCP server."""
return await self.ap.tool_mgr.mcp_tool_loader.get_resource_templates(server_name)
async def read_mcp_server_resource_envelope(
self,
server_name: str,
uri: str,
*,
max_bytes: int | None = None,
include_blob: bool = False,
) -> dict:
"""Read a resource from a specific MCP server with metadata."""
kwargs = {'include_blob': include_blob, 'source': 'ui_preview'}
if max_bytes is not None:
kwargs['max_bytes'] = max_bytes
return await self.ap.tool_mgr.mcp_tool_loader.read_resource_envelope(server_name, uri, **kwargs)
async def read_mcp_server_resource(self, server_name: str, uri: str) -> list[dict]:
"""Read a resource from a specific MCP server."""
return await self.ap.tool_mgr.mcp_tool_loader.read_resource(server_name, uri)
async def test_mcp_server(self, server_name: str, server_data: dict) -> int:
"""测试 MCP 服务器连接并返回任务 ID"""
src/langbot/pkg/api/http/service/pipeline.py
+10
View File
@@ -185,6 +185,8 @@ class PipelineService:
'enable_all_mcp_servers': True,
'plugins': [],
'mcp_servers': [],
'mcp_resources': [],
'mcp_resource_agent_read_enabled': True,
}
await self.ap.persistence_mgr.execute_async(
@@ -284,6 +286,8 @@ class PipelineService:
'enable_all_mcp_servers': True,
'plugins': [],
'mcp_servers': [],
'mcp_resources': [],
'mcp_resource_agent_read_enabled': True,
}
),
}
@@ -308,6 +312,8 @@ class PipelineService:
enable_all_mcp_servers: bool = True,
bound_skills: list[str] = None,
enable_all_skills: bool = True,
bound_mcp_resources: list[dict] = None,
mcp_resource_agent_read_enabled: bool | None = None,
) -> None:
"""Update the bound plugins and MCP servers for a pipeline"""
# Get current pipeline
@@ -327,10 +333,14 @@ class PipelineService:
extensions_preferences['enable_all_mcp_servers'] = enable_all_mcp_servers
extensions_preferences['enable_all_skills'] = enable_all_skills
extensions_preferences['plugins'] = bound_plugins
if mcp_resource_agent_read_enabled is not None:
extensions_preferences['mcp_resource_agent_read_enabled'] = mcp_resource_agent_read_enabled
if bound_mcp_servers is not None:
extensions_preferences['mcp_servers'] = bound_mcp_servers
if bound_skills is not None:
extensions_preferences['skills'] = bound_skills
if bound_mcp_resources is not None:
extensions_preferences['mcp_resources'] = bound_mcp_resources
await self.ap.persistence_mgr.execute_async(
sqlalchemy.update(persistence_pipeline.LegacyPipeline)
src/langbot/pkg/api/http/service/user.py
+16 -16
View File
@@ -20,6 +20,15 @@ class UserService:
def __init__(self, ap: app.Application) -> None:
self.ap = ap
self._create_user_lock = asyncio.Lock()
self._password_hash_lock = asyncio.Semaphore(1)
async def _hash_password(self, password: str) -> str:
async with self._password_hash_lock:
return await asyncio.to_thread(argon2.PasswordHasher().hash, password)
async def _verify_password(self, hashed_password: str, password: str) -> None:
async with self._password_hash_lock:
await asyncio.to_thread(argon2.PasswordHasher().verify, hashed_password, password)
async def is_initialized(self) -> bool:
result = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(user.User).limit(1))
@@ -28,9 +37,7 @@ class UserService:
return result_list is not None and len(result_list) > 0
async def create_user(self, user_email: str, password: str) -> None:
ph = argon2.PasswordHasher()
hashed_password = ph.hash(password)
hashed_password = await self._hash_password(password)
await self.ap.persistence_mgr.execute_async(
sqlalchemy.insert(user.User).values(user=user_email, password=hashed_password, account_type='local')
@@ -69,9 +76,7 @@ class UserService:
if not user_obj.password:
raise ValueError('请使用 Space 账户登录')
ph = argon2.PasswordHasher()
ph.verify(user_obj.password, password)
await self._verify_password(user_obj.password, password)
return await self.generate_jwt_token(user_email)
@@ -93,17 +98,13 @@ class UserService:
return jwt.decode(token, jwt_secret, algorithms=['HS256'])['user']
async def reset_password(self, user_email: str, new_password: str) -> None:
ph = argon2.PasswordHasher()
hashed_password = ph.hash(new_password)
hashed_password = await self._hash_password(new_password)
await self.ap.persistence_mgr.execute_async(
sqlalchemy.update(user.User).where(user.User.user == user_email).values(password=hashed_password)
)
async def change_password(self, user_email: str, current_password: str, new_password: str) -> None:
ph = argon2.PasswordHasher()
user_obj = await self.get_user_by_email(user_email)
if user_obj is None:
raise ValueError('User not found')
@@ -111,9 +112,9 @@ class UserService:
if not user_obj.password:
raise ValueError('No local password set, please set a password first')
ph.verify(user_obj.password, current_password)
await self._verify_password(user_obj.password, current_password)
hashed_password = ph.hash(new_password)
hashed_password = await self._hash_password(new_password)
await self.ap.persistence_mgr.execute_async(
sqlalchemy.update(user.User).where(user.User.user == user_email).values(password=hashed_password)
@@ -232,7 +233,6 @@ class UserService:
async def set_password(self, user_email: str, new_password: str, current_password: str | None = None) -> None:
"""Set or change password for a user"""
ph = argon2.PasswordHasher()
user_obj = await self.get_user_by_email(user_email)
if user_obj is None:
@@ -243,9 +243,9 @@ class UserService:
if has_password:
if not current_password:
raise ValueError('Current password is required')
ph.verify(user_obj.password, current_password)
await self._verify_password(user_obj.password, current_password)
hashed_password = ph.hash(new_password)
hashed_password = await self._hash_password(new_password)
await self.ap.persistence_mgr.execute_async(
sqlalchemy.update(user.User).where(user.User.user == user_email).values(password=hashed_password)
)
src/langbot/pkg/box/service.py
+5 -2
View File
@@ -82,7 +82,6 @@ class BoxService:
return self._enabled
async def initialize(self):
self._ensure_default_workspace()
if not self._enabled:
# Disabled by config: do NOT connect to a remote runtime, do NOT
# fork a stdio subprocess. Every consumer of box_service should
@@ -99,6 +98,7 @@ class BoxService:
await self._runtime_connector.initialize()
else:
await self.client.initialize()
self._ensure_default_workspace()
self._available = True
self._connector_error = ''
self.ap.logger.info(
@@ -1158,6 +1158,9 @@ class BoxService:
if self.default_workspace is None:
return
if not self.shares_filesystem_with_box:
return
if os.path.isdir(self.default_workspace):
return
@@ -1182,7 +1185,7 @@ class BoxService:
return
host_path = os.path.realpath(spec.host_path)
if not os.path.isdir(host_path):
if self.shares_filesystem_with_box and not os.path.isdir(host_path):
raise BoxValidationError('host_path must point to an existing directory on the host')
if not self.allowed_mount_roots:
src/langbot/pkg/command/cmdmgr.py
+11 -1
View File
@@ -84,7 +84,17 @@ class CommandManager:
privilege = 1
if f'{query.launcher_type.value}_{query.launcher_id}' in self.ap.instance_config.data['admins']:
import sqlalchemy as _sa
from ..entity.persistence.bot import BotAdmin as _BotAdmin
_admins = await self.ap.persistence_mgr.execute_async(
_sa.select(_BotAdmin).where(
_BotAdmin.bot_uuid == (query.bot_uuid or ''),
_BotAdmin.launcher_type == query.launcher_type.value,
_BotAdmin.launcher_id == str(query.launcher_id),
)
)
if _admins.first() is not None:
privilege = 2
ctx = command_context.ExecuteContext(
src/langbot/pkg/entity/persistence/bot.py
+14
View File
@@ -3,6 +3,20 @@ import sqlalchemy
from .base import Base
class BotAdmin(Base):
"""Bot admin — a launcher that has admin privilege for a specific bot's commands"""
__tablename__ = 'bot_admins'
id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True, autoincrement=True)
bot_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
launcher_type = sqlalchemy.Column(sqlalchemy.String(64), nullable=False)
launcher_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())
__table_args__ = (sqlalchemy.UniqueConstraint('bot_uuid', 'launcher_type', 'launcher_id', name='uq_bot_admin'),)
class Bot(Base):
"""Bot"""
src/langbot/pkg/entity/persistence/pipeline.py
+8 -1
View File
@@ -26,7 +26,14 @@ class LegacyPipeline(Base):
extensions_preferences = sqlalchemy.Column(
sqlalchemy.JSON,
nullable=False,
default={'enable_all_plugins': True, 'enable_all_mcp_servers': True, 'plugins': [], 'mcp_servers': []},
default={
'enable_all_plugins': True,
'enable_all_mcp_servers': True,
'plugins': [],
'mcp_servers': [],
'mcp_resources': [],
'mcp_resource_agent_read_enabled': True,
},
)
src/langbot/pkg/persistence/alembic/versions/0007_add_bot_admins.py
+84
View File
@@ -0,0 +1,84 @@
"""add bot_admins table and migrate config admins
Revision ID: 0007_add_bot_admins
Revises: 0006_normalize_mcp_remote_mode
Create Date: 2026-06-26
"""
import sqlalchemy as sa
from alembic import op
revision = '0007_add_bot_admins'
down_revision = '0006_normalize_mcp_remote_mode'
branch_labels = None
depends_on = None
def upgrade() -> None:
conn = op.get_bind()
if 'bot_admins' in sa.inspect(conn).get_table_names():
return
op.create_table(
'bot_admins',
sa.Column('id', sa.Integer, primary_key=True, autoincrement=True),
sa.Column('bot_uuid', sa.String(255), nullable=False),
sa.Column('launcher_type', sa.String(64), nullable=False),
sa.Column('launcher_id', sa.String(255), nullable=False),
sa.Column('created_at', sa.DateTime, nullable=False, server_default=sa.func.now()),
sa.UniqueConstraint('bot_uuid', 'launcher_type', 'launcher_id', name='uq_bot_admin'),
)
# Migrate old config-based admins into the first bot (best-effort)
inspector = sa.inspect(conn)
tables = inspector.get_table_names()
if 'bots' not in tables:
return
# Read the first bot uuid
row = conn.execute(sa.text('SELECT uuid FROM bots ORDER BY created_at LIMIT 1')).first()
if row is None:
return
first_bot_uuid = row[0]
# Read instance_config metadata key that holds the admins list
if 'metadata' not in tables:
return
meta_row = conn.execute(sa.text("SELECT value FROM metadata WHERE key = 'instance_config'")).first()
if meta_row is None:
return
import json
try:
cfg = json.loads(meta_row[0])
except Exception:
return
admins = cfg.get('admins', [])
for entry in admins:
parts = entry.split('_', 1)
if len(parts) != 2:
continue
launcher_type, launcher_id = parts
try:
conn.execute(
sa.text(
'INSERT OR IGNORE INTO bot_admins (bot_uuid, launcher_type, launcher_id) VALUES (:bu, :lt, :li)'
),
{'bu': first_bot_uuid, 'lt': launcher_type, 'li': launcher_id},
)
except Exception:
pass
# Remove admins key from stored config
if 'admins' in cfg:
del cfg['admins']
conn.execute(
sa.text("UPDATE metadata SET value = :v WHERE key = 'instance_config'"),
{'v': json.dumps(cfg)},
)
def downgrade() -> None:
op.drop_table('bot_admins')
src/langbot/pkg/persistence/alembic/versions/0008_add_mcp_resource_preferences.py
+95
View File
@@ -0,0 +1,95 @@
"""add mcp resource preferences to pipelines
Revision ID: 0008_mcp_resource_prefs
Revises: 0007_add_bot_admins
Create Date: 2026-06-30
"""
from __future__ import annotations
import json
from typing import Any
import sqlalchemy as sa
from alembic import op
revision = '0008_mcp_resource_prefs'
down_revision = '0007_add_bot_admins'
branch_labels = None
depends_on = None
_PIPELINE_TABLE = sa.table(
'legacy_pipelines',
sa.column('uuid', sa.String(255)),
sa.column('extensions_preferences', sa.JSON()),
)
def _has_extensions_preferences_table(conn: sa.Connection) -> bool:
inspector = sa.inspect(conn)
if 'legacy_pipelines' not in inspector.get_table_names():
return False
columns = {column['name'] for column in inspector.get_columns('legacy_pipelines')}
return 'extensions_preferences' in columns
def _decode_preferences(value: Any) -> dict[str, Any]:
if value is None:
return {}
if isinstance(value, dict):
return dict(value)
if isinstance(value, str):
try:
decoded = json.loads(value)
except json.JSONDecodeError:
return {}
if isinstance(decoded, dict):
return decoded
return {}
def _update_preferences(conn: sa.Connection, uuid: str, preferences: dict[str, Any]) -> None:
conn.execute(
_PIPELINE_TABLE.update().where(_PIPELINE_TABLE.c.uuid == uuid).values(extensions_preferences=preferences)
)
def upgrade() -> None:
conn = op.get_bind()
if not _has_extensions_preferences_table(conn):
return
rows = conn.execute(sa.select(_PIPELINE_TABLE.c.uuid, _PIPELINE_TABLE.c.extensions_preferences)).all()
for uuid, raw_preferences in rows:
preferences = _decode_preferences(raw_preferences)
changed = False
if 'mcp_resources' not in preferences:
preferences['mcp_resources'] = []
changed = True
if 'mcp_resource_agent_read_enabled' not in preferences:
preferences['mcp_resource_agent_read_enabled'] = True
changed = True
if changed:
_update_preferences(conn, uuid, preferences)
def downgrade() -> None:
conn = op.get_bind()
if not _has_extensions_preferences_table(conn):
return
rows = conn.execute(sa.select(_PIPELINE_TABLE.c.uuid, _PIPELINE_TABLE.c.extensions_preferences)).all()
for uuid, raw_preferences in rows:
preferences = _decode_preferences(raw_preferences)
changed = False
for key in ('mcp_resources', 'mcp_resource_agent_read_enabled'):
if key in preferences:
preferences.pop(key)
changed = True
if changed:
_update_preferences(conn, uuid, preferences)
src/langbot/pkg/pipeline/monitoring_helper.py
+4 -4
View File
@@ -32,7 +32,7 @@ class MonitoringHelper:
"""Record the start of query processing, returns message_id"""
try:
# Check if session exists, if not, record session start
session_id = f'{query.launcher_type}_{query.launcher_id}'
session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'
# Get sender name from message event
sender_name = None
@@ -137,7 +137,7 @@ class MonitoringHelper:
):
"""Record bot response message to monitoring"""
try:
session_id = f'{query.launcher_type}_{query.launcher_id}'
session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'
# Get sender name from message event
sender_name = None
@@ -202,7 +202,7 @@ class MonitoringHelper:
) -> str:
"""Record query processing error, returns message_id"""
try:
session_id = f'{query.launcher_type}_{query.launcher_id}'
session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'
# Get sender name from message event
sender_name = None
@@ -268,7 +268,7 @@ class MonitoringHelper:
):
"""Record LLM call"""
try:
session_id = f'{query.launcher_type}_{query.launcher_id}'
session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'
await ap.monitoring_service.record_llm_call(
bot_id=bot_id,
src/langbot/pkg/pipeline/pipelinemgr.py
+32 -1
View File
@@ -14,6 +14,7 @@ import langbot_plugin.api.entities.builtin.platform.events as platform_events
import langbot_plugin.api.entities.events as events
from ..utils import importutil
from .config_coercion import coerce_pipeline_config
from ..agent.runner.config_migration import ConfigMigration
import langbot_plugin.api.entities.builtin.provider.session as provider_session
import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
@@ -95,6 +96,34 @@ class RuntimePipeline:
self.enable_all_plugins = extensions_prefs.get('enable_all_plugins', True)
self.enable_all_mcp_servers = extensions_prefs.get('enable_all_mcp_servers', True)
pipeline_config = pipeline_entity.config or {}
ai_config = pipeline_config.get('ai', {}) if isinstance(pipeline_config, dict) else {}
legacy_local_agent_config = ai_config.get('local-agent', {}) if isinstance(ai_config, dict) else {}
if not isinstance(legacy_local_agent_config, dict):
legacy_local_agent_config = {}
runner_config: dict[str, typing.Any] = {}
runner_id = ConfigMigration.resolve_runner_id(pipeline_config) if isinstance(pipeline_config, dict) else None
if runner_id:
resolved_runner_config = ConfigMigration.resolve_runner_config(pipeline_config, runner_id)
if isinstance(resolved_runner_config, dict):
runner_config = resolved_runner_config
self.mcp_resource_attachments = runner_config.get(
'mcp-resources',
legacy_local_agent_config.get(
'mcp-resources',
extensions_prefs.get('mcp_resources', []),
),
)
self.mcp_resource_agent_read_enabled = runner_config.get(
'mcp-resource-agent-read-enabled',
legacy_local_agent_config.get(
'mcp-resource-agent-read-enabled',
extensions_prefs.get('mcp_resource_agent_read_enabled', True),
),
)
if self.enable_all_plugins:
# None indicates to use all available plugins
self.bound_plugins = None
@@ -114,6 +143,8 @@ class RuntimePipeline:
# Store bound plugins and MCP servers in query for filtering
query.variables['_pipeline_bound_plugins'] = self.bound_plugins
query.variables['_pipeline_bound_mcp_servers'] = self.bound_mcp_servers
query.variables['_pipeline_mcp_resource_attachments'] = self.mcp_resource_attachments
query.variables['_pipeline_mcp_resource_agent_read_enabled'] = self.mcp_resource_agent_read_enabled
# Record query start for monitoring
try:
@@ -176,7 +207,7 @@ class RuntimePipeline:
bot_name = query.variables.get('_monitoring_bot_name', 'Unknown')
pipeline_name = query.variables.get('_monitoring_pipeline_name', 'Unknown')
message_id = query.variables.get('_monitoring_message_id', '')
session_id = f'{query.launcher_type}_{query.launcher_id}'
session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'
# Update message status to error
if message_id:
src/langbot/pkg/pipeline/plugin_diagnostics.py
+274
View File
@@ -0,0 +1,274 @@
from __future__ import annotations
import traceback
import weakref
from dataclasses import dataclass, field
from typing import Any
import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
import langbot_plugin.api.entities.builtin.platform.message as platform_message
@dataclass(frozen=True)
class PluginResponseSource:
plugin: dict[str, str]
event_name: str | None = None
is_approximate: bool = False
@dataclass
class QueryDiagnosticState:
pending_by_chain_id: dict[int, list[PluginResponseSource]] = field(default_factory=dict)
by_response_index: dict[int, list[PluginResponseSource]] = field(default_factory=dict)
finalizer: weakref.finalize | None = None
_QUERY_STATES: dict[int, QueryDiagnosticState] = {}
def record_plugin_response_source(
query: pipeline_query.Query,
response_index: int,
response_sources: list[dict[str, Any]] | None,
emitted_plugins: list[dict[str, Any]] | None = None,
event_name: str | None = None,
) -> None:
plugin_sources = _build_plugin_sources(response_sources, emitted_plugins, event_name)
if not plugin_sources:
return
state = _get_or_create_query_state(query)
state.by_response_index[response_index] = plugin_sources
def record_last_plugin_response_source(
query: pipeline_query.Query,
response_sources: list[dict[str, Any]] | None,
emitted_plugins: list[dict[str, Any]] | None = None,
event_name: str | None = None,
) -> None:
record_plugin_response_source(
query,
len(query.resp_message_chain) - 1,
response_sources,
emitted_plugins,
event_name,
)
def record_pending_plugin_response_source(
query: pipeline_query.Query,
message_chain: platform_message.MessageChain,
response_sources: list[dict[str, Any]] | None,
emitted_plugins: list[dict[str, Any]] | None = None,
event_name: str | None = None,
) -> None:
plugin_sources = _build_plugin_sources(response_sources, emitted_plugins, event_name)
if not plugin_sources:
return
state = _get_or_create_query_state(query)
state.pending_by_chain_id[id(message_chain)] = plugin_sources
def consume_pending_plugin_response_source(
query: pipeline_query.Query,
message_chain: platform_message.MessageChain,
response_index: int,
) -> None:
state = _get_query_state(query)
if state is None:
return
source = state.pending_by_chain_id.pop(id(message_chain), None)
if source is None:
return
state.by_response_index[response_index] = source
def clear_response_source(query: pipeline_query.Query, response_index: int) -> None:
state = _get_query_state(query)
if state is None:
return
state.by_response_index.pop(response_index, None)
_discard_query_state_if_empty(query)
async def notify_response_delivery_failure(
ap: Any,
query: pipeline_query.Query,
response_index: int,
message_chain: platform_message.MessageChain,
error: Exception,
) -> None:
try:
plugin_refs = _get_response_sources(query, response_index)
if not plugin_refs:
return
connector = getattr(ap, 'plugin_connector', None)
if connector is None or not hasattr(connector, 'notify_plugin_diagnostic'):
return
for source in plugin_refs:
payload = _build_delivery_failure_payload(
plugin_ref=source.plugin,
event_name=source.event_name,
is_approximate=source.is_approximate,
query=query,
response_index=response_index,
message_chain=message_chain,
error=error,
)
try:
await connector.notify_plugin_diagnostic(payload)
except Exception as diag_error:
_debug(ap, f'Plugin diagnostic forwarding failed: {diag_error}')
except Exception as diag_error:
_debug(ap, f'Plugin diagnostic forwarding skipped: {diag_error}')
def get_emitted_plugins(event_ctx: Any) -> list[dict[str, Any]]:
emitted_plugins = getattr(event_ctx, '_emitted_plugins', [])
return emitted_plugins if isinstance(emitted_plugins, list) else []
def get_response_sources(event_ctx: Any) -> list[dict[str, Any]] | None:
event_attrs = vars(event_ctx)
if '_response_sources' not in event_attrs:
return None
response_sources = event_attrs['_response_sources']
return response_sources if isinstance(response_sources, list) else []
def _get_or_create_query_state(query: pipeline_query.Query) -> QueryDiagnosticState:
query_key = id(query)
state = _QUERY_STATES.get(query_key)
if state is not None:
return state
state = QueryDiagnosticState()
try:
state.finalizer = weakref.finalize(query, _discard_query_state, query_key)
except TypeError:
state.finalizer = None
_QUERY_STATES[query_key] = state
return state
def _get_query_state(query: pipeline_query.Query) -> QueryDiagnosticState | None:
return _QUERY_STATES.get(id(query))
def _discard_query_state(query_key: int) -> None:
_QUERY_STATES.pop(query_key, None)
def _discard_query_state_if_empty(query: pipeline_query.Query) -> None:
query_key = id(query)
state = _QUERY_STATES.get(query_key)
if state is None:
return
if state.pending_by_chain_id or state.by_response_index:
return
if state.finalizer is not None:
state.finalizer.detach()
_discard_query_state(query_key)
def _get_response_sources(
query: pipeline_query.Query,
response_index: int,
) -> list[PluginResponseSource]:
state = _get_query_state(query)
if state is None:
return []
return state.by_response_index.get(response_index, [])
def _extract_plugin_ref(plugin: Any) -> dict[str, str] | None:
manifest = plugin.get('manifest') if isinstance(plugin, dict) else None
metadata = manifest.get('metadata') if isinstance(manifest, dict) else None
if not isinstance(metadata, dict):
return None
author = metadata.get('author')
name = metadata.get('name')
if not author or not name:
return None
return {'author': str(author), 'name': str(name)}
def _extract_response_source_plugin_ref(source: Any) -> dict[str, str] | None:
if not isinstance(source, dict):
return None
if source.get('kind') != 'reply_message_chain':
return None
plugin_ref = source.get('plugin')
if not isinstance(plugin_ref, dict):
return None
author = plugin_ref.get('author')
name = plugin_ref.get('name')
if not author or not name:
return None
return {'author': str(author), 'name': str(name)}
def _build_plugin_sources(
response_sources: list[dict[str, Any]] | None,
emitted_plugins: list[dict[str, Any]] | None,
event_name: str | None,
) -> list[PluginResponseSource]:
if response_sources is not None:
plugin_refs = [_extract_response_source_plugin_ref(source) for source in response_sources]
return [
PluginResponseSource(plugin=plugin, event_name=event_name) for plugin in plugin_refs if plugin is not None
]
if emitted_plugins:
plugin_refs = [_extract_plugin_ref(plugin) for plugin in emitted_plugins]
return [
PluginResponseSource(plugin=plugin, event_name=event_name, is_approximate=True)
for plugin in plugin_refs
if plugin is not None
]
return []
def _debug(ap: Any, message: str) -> None:
logger = getattr(ap, 'logger', None)
if logger is not None:
logger.debug(message)
def _build_delivery_failure_payload(
plugin_ref: dict[str, str],
event_name: str | None,
is_approximate: bool,
query: pipeline_query.Query,
response_index: int,
message_chain: platform_message.MessageChain,
error: Exception,
) -> dict[str, Any]:
details: dict[str, Any] = {
'message_component_types': [component.__class__.__name__ for component in message_chain],
'message_preview': str(message_chain)[:200],
}
if is_approximate:
details['attribution_warning'] = (
'This diagnostic was delivered to all plugins that handled the event because the '
'plugin runtime did not report the exact reply_message_chain source.'
)
return {
'level': 'ERROR',
'code': 'response_delivery_failed',
'message': 'Failed to deliver a plugin-provided response message.',
'plugin': plugin_ref,
'query': {
'query_id': query.query_id,
'event_name': event_name or query.message_event.__class__.__name__,
'stage': query.current_stage_name or 'SendResponseBackStage',
'response_index': response_index,
},
'details': details,
'delivery': {
'error_type': error.__class__.__name__,
'error_message': str(error),
'traceback': traceback.format_exception_only(type(error), error)[-1].strip(),
},
}
src/langbot/pkg/pipeline/preproc/preproc.py
+26 -6
View File
@@ -136,9 +136,7 @@ class PreProcessor(stage.PipelineStage):
strict_thread=True,
)
except Exception as e:
self.ap.logger.warning(
f'Unable to load Transcript history view for conversation {conversation_uuid}: {e}'
)
self.ap.logger.warning(f'Unable to load Transcript history view for conversation {conversation_uuid}: {e}')
return None
return messages or None
@@ -161,6 +159,21 @@ class PreProcessor(stage.PipelineStage):
return transcript_messages
return conversation.messages.copy()
@staticmethod
def _filter_selected_tools(
tools: list,
runner_config: dict,
) -> list:
if runner_config.get('enable-all-tools', True) is not False:
return tools
selected_tools = runner_config.get('tools', [])
if not isinstance(selected_tools, list):
return []
selected_tool_names = {tool for tool in selected_tools if isinstance(tool, str)}
return [tool for tool in tools if tool.name in selected_tool_names]
async def process(
self,
query: pipeline_query.Query,
@@ -181,6 +194,7 @@ class PreProcessor(stage.PipelineStage):
uses_host_models = config_schema.uses_host_models(descriptor)
uses_host_tools = config_schema.uses_host_tools(descriptor)
include_mcp_resource_tools = query.variables.get('_pipeline_mcp_resource_agent_read_enabled', True)
llm_model = None
if uses_host_models:
primary_uuid, fallback_uuids = config_schema.extract_model_selection(descriptor, runner_config)
@@ -235,10 +249,12 @@ class PreProcessor(stage.PipelineStage):
query.use_llm_model_uuid = llm_model.model_entity.uuid
if uses_host_tools and 'func_call' in (llm_model.model_entity.abilities or []):
query.use_funcs = await self.ap.tool_mgr.get_all_tools(
all_tools = await self.ap.tool_mgr.get_all_tools(
bound_plugins,
bound_mcp_servers,
include_mcp_resource_tools=include_mcp_resource_tools,
)
query.use_funcs = self._filter_selected_tools(all_tools, runner_config)
self.ap.logger.debug(f'Bound plugins: {bound_plugins}')
self.ap.logger.debug(f'Bound MCP servers: {bound_mcp_servers}')
@@ -247,15 +263,19 @@ class PreProcessor(stage.PipelineStage):
# If primary model doesn't support func_call but fallback models exist,
# load tools anyway since fallback models may support them
if uses_host_tools and not query.use_funcs and query.variables.get('_fallback_model_uuids'):
query.use_funcs = await self.ap.tool_mgr.get_all_tools(
all_tools = await self.ap.tool_mgr.get_all_tools(
bound_plugins,
bound_mcp_servers,
include_mcp_resource_tools=include_mcp_resource_tools,
)
query.use_funcs = self._filter_selected_tools(all_tools, runner_config)
elif uses_host_tools:
query.use_funcs = await self.ap.tool_mgr.get_all_tools(
all_tools = await self.ap.tool_mgr.get_all_tools(
bound_plugins,
bound_mcp_servers,
include_mcp_resource_tools=include_mcp_resource_tools,
)
query.use_funcs = self._filter_selected_tools(all_tools, runner_config)
self.ap.logger.debug(f'Bound plugins: {bound_plugins}')
self.ap.logger.debug(f'Bound MCP servers: {bound_mcp_servers}')
src/langbot/pkg/pipeline/process/handlers/chat.py
+8
View File
@@ -9,6 +9,7 @@ from datetime import datetime
from .. import handler
from ... import entities
from ... import plugin_diagnostics
import langbot_plugin.api.entities.events as events
from ....agent.runner.config_migration import ConfigMigration
@@ -64,6 +65,13 @@ class ChatMessageHandler(handler.MessageHandler):
if event_ctx.is_prevented_default():
if event_ctx.event.reply_message_chain is not None:
mc = event_ctx.event.reply_message_chain
plugin_diagnostics.record_pending_plugin_response_source(
query,
mc,
plugin_diagnostics.get_response_sources(event_ctx),
plugin_diagnostics.get_emitted_plugins(event_ctx),
event.event_name,
)
query.resp_messages.append(mc)
yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
src/langbot/pkg/pipeline/process/handlers/command.py
+8
View File
@@ -4,6 +4,7 @@ import typing
from .. import handler
from ... import entities
from ... import plugin_diagnostics
import langbot_plugin.api.entities.builtin.provider.message as provider_message
import langbot_plugin.api.entities.builtin.provider.session as provider_session
import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
@@ -52,6 +53,13 @@ class CommandHandler(handler.MessageHandler):
if event_ctx.is_prevented_default():
if event_ctx.event.reply_message_chain is not None:
mc = event_ctx.event.reply_message_chain
plugin_diagnostics.record_pending_plugin_response_source(
query,
mc,
plugin_diagnostics.get_response_sources(event_ctx),
plugin_diagnostics.get_emitted_plugins(event_ctx),
event.event_name,
)
query.resp_messages.append(mc)
src/langbot/pkg/pipeline/respback/respback.py
+30 -14
View File
@@ -9,6 +9,7 @@ import langbot_plugin.api.entities.builtin.platform.message as platform_message
import langbot_plugin.api.entities.builtin.provider.message as provider_message
from .. import stage, entities
from .. import plugin_diagnostics
import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
@@ -39,20 +40,35 @@ class SendResponseBackStage(stage.PipelineStage):
has_chunks = any(isinstance(msg, provider_message.MessageChunk) for msg in query.resp_messages)
# TODO 命令与流式的兼容性问题
if await query.adapter.is_stream_output_supported() and has_chunks:
is_final = [msg.is_final for msg in query.resp_messages][0]
await query.adapter.reply_message_chunk(
message_source=query.message_event,
bot_message=query.resp_messages[-1],
message=query.resp_message_chain[-1],
quote_origin=quote_origin,
is_final=is_final,
)
else:
await query.adapter.reply_message(
message_source=query.message_event,
message=query.resp_message_chain[-1],
quote_origin=quote_origin,
response_index = len(query.resp_message_chain) - 1
message_chain = query.resp_message_chain[-1]
try:
if await query.adapter.is_stream_output_supported() and has_chunks:
is_final = [msg.is_final for msg in query.resp_messages][0]
await query.adapter.reply_message_chunk(
message_source=query.message_event,
bot_message=query.resp_messages[-1],
message=message_chain,
quote_origin=quote_origin,
is_final=is_final,
)
else:
await query.adapter.reply_message(
message_source=query.message_event,
message=message_chain,
quote_origin=quote_origin,
)
except Exception as e:
await plugin_diagnostics.notify_response_delivery_failure(
self.ap,
query,
response_index,
message_chain,
e,
)
plugin_diagnostics.clear_response_source(query, response_index)
raise
plugin_diagnostics.clear_response_source(query, response_index)
return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
src/langbot/pkg/pipeline/wrapper/wrapper.py
+21
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import typing
from .. import entities
from .. import plugin_diagnostics
from .. import stage
import langbot_plugin.api.entities.builtin.platform.message as platform_message
@@ -78,6 +79,11 @@ class ResponseWrapper(stage.PipelineStage):
# 如果 resp_messages[-1] 已经是 MessageChain 了
if isinstance(query.resp_messages[-1], platform_message.MessageChain):
query.resp_message_chain.append(query.resp_messages[-1])
plugin_diagnostics.consume_pending_plugin_response_source(
query,
query.resp_messages[-1],
len(query.resp_message_chain) - 1,
)
yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
@@ -129,8 +135,10 @@ class ResponseWrapper(stage.PipelineStage):
else:
if event_ctx.event.reply_message_chain is not None:
reply_chain = event_ctx.event.reply_message_chain
is_plugin_reply = True
else:
reply_chain = result.get_content_platform_message_chain()
is_plugin_reply = False
# Attach files the agent produced in the sandbox
# outbox, but only on the terminal assistant message.
@@ -138,6 +146,13 @@ class ResponseWrapper(stage.PipelineStage):
await self._append_outbound_attachments(query, reply_chain)
query.resp_message_chain.append(reply_chain)
if is_plugin_reply:
plugin_diagnostics.record_last_plugin_response_source(
query,
plugin_diagnostics.get_response_sources(event_ctx),
plugin_diagnostics.get_emitted_plugins(event_ctx),
event.event_name,
)
yield entities.StageProcessResult(
result_type=entities.ResultType.CONTINUE,
@@ -180,6 +195,12 @@ class ResponseWrapper(stage.PipelineStage):
else:
if event_ctx.event.reply_message_chain is not None:
query.resp_message_chain.append(event_ctx.event.reply_message_chain)
plugin_diagnostics.record_last_plugin_response_source(
query,
plugin_diagnostics.get_response_sources(event_ctx),
plugin_diagnostics.get_emitted_plugins(event_ctx),
event.event_name,
)
else:
query.resp_message_chain.append(
src/langbot/pkg/platform/sources/aiocqhttp.py
+15 -6
View File
@@ -16,6 +16,14 @@ from ...utils import image
import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger
def _normalize_base64_payload(value: str) -> str:
if value.startswith('base64://'):
return value.removeprefix('base64://')
if value.startswith('data:') and ';base64,' in value:
return value.split(';base64,', 1)[1]
return value
class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
@staticmethod
async def yiri2target(
@@ -35,7 +43,7 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
elif type(msg) is platform_message.Image:
arg = ''
if msg.base64:
arg = msg.base64
arg = _normalize_base64_payload(msg.base64)
msg_list.append(aiocqhttp.MessageSegment.image(f'base64://{arg}'))
elif msg.url:
arg = msg.url
@@ -50,7 +58,7 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
elif type(msg) is platform_message.Voice:
arg = ''
if msg.base64:
arg = msg.base64
arg = _normalize_base64_payload(msg.base64)
msg_list.append(aiocqhttp.MessageSegment.record(f'base64://{arg}'))
elif msg.url:
arg = msg.url
@@ -62,7 +70,10 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
for node in msg.node_list:
msg_list.extend((await AiocqhttpMessageConverter.yiri2target(node.message_chain))[0])
elif isinstance(msg, platform_message.File):
msg_list.append({'type': 'file', 'data': {'file': msg.url, 'name': msg.name}})
file = msg.url or msg.path
if not file and msg.base64:
file = f'base64://{_normalize_base64_payload(msg.base64)}'
msg_list.append({'type': 'file', 'data': {'file': file, 'name': msg.name}})
elif isinstance(msg, platform_message.Face):
if msg.face_type == 'face':
msg_list.append(aiocqhttp.MessageSegment.face(msg.face_id))
@@ -433,9 +444,7 @@ class AiocqhttpAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
elif isinstance(component, platform_message.Image):
img_data = {}
if component.base64:
b64 = component.base64
if b64.startswith('data:'):
b64 = b64.split(',', 1)[-1] if ',' in b64 else b64
b64 = _normalize_base64_payload(component.base64)
img_data['file'] = f'base64://{b64}'
elif component.url:
img_data['file'] = component.url
src/langbot/pkg/plugin/connector.py
+14
View File
@@ -751,6 +751,8 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
event_ctx = context.EventContext.from_event(event)
if not self.is_enable_plugin:
event_ctx._emitted_plugins = []
event_ctx._response_sources = []
return event_ctx
# Pass include_plugins to runtime for filtering
@@ -759,9 +761,21 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
)
event_ctx = context.EventContext.model_validate(event_ctx_result['event_context'])
event_ctx._emitted_plugins = event_ctx_result.get('emitted_plugins', [])
if 'response_sources' in event_ctx_result:
event_ctx._response_sources = event_ctx_result['response_sources']
return event_ctx
async def notify_plugin_diagnostic(self, diagnostic: dict[str, Any]) -> None:
"""Best-effort diagnostic forwarding to the plugin runtime."""
if not self.is_enable_plugin:
return
try:
await self.handler.notify_plugin_diagnostic(diagnostic)
except Exception as e:
self.ap.logger.debug(f'Plugin diagnostic forwarding skipped: {e}')
async def list_tools(self, bound_plugins: list[str] | None = None) -> list[ComponentManifest]:
if not self.is_enable_plugin:
return []
src/langbot/pkg/plugin/handler.py
+21
View File
@@ -51,6 +51,15 @@ def _serialize_plugin_api_result(value: Any) -> Any:
return value
class _RawAction:
def __init__(self, value: str):
self.value = value
def _langbot_to_runtime_action(enum_name: str, fallback_value: str) -> Any:
return getattr(LangBotToRuntimeAction, enum_name, _RawAction(fallback_value))
def _make_rag_error_response(error: Exception, error_type: str, **extra_context) -> handler.ActionResponse:
"""Create a clean error response for RAG operations.
@@ -1541,6 +1550,18 @@ class RuntimeConnectionHandler(handler.Handler):
return result
async def notify_plugin_diagnostic(self, diagnostic: dict[str, Any]) -> dict[str, Any]:
"""Notify the plugin runtime about a best-effort plugin diagnostic.
This intentionally uses the raw protocol string instead of a SDK enum so
LangBot can keep running with older langbot-plugin versions.
"""
return await self.call_action(
_langbot_to_runtime_action('PLUGIN_DIAGNOSTIC', 'plugin_diagnostic'),
diagnostic,
timeout=5,
)
async def list_tools(self, include_plugins: list[str] | None = None) -> list[dict[str, Any]]:
"""List tools"""
result = await self.call_action(
src/langbot/pkg/provider/tools/loaders/mcp.py
+875 -26
View File
File diff suppressed because it is too large Load Diff
src/langbot/pkg/provider/tools/loaders/plugin.py
+18
View File
@@ -32,6 +32,24 @@ class PluginToolLoader(loader.ToolLoader):
return all_functions
async def get_tool_catalog(self, bound_plugins: list[str] | None = None) -> list[dict[str, typing.Any]]:
catalog: list[dict[str, typing.Any]] = []
for tool in await self.ap.plugin_connector.list_tools(bound_plugins):
catalog.append(
{
'name': tool.metadata.name,
'description': tool.spec['llm_prompt'],
'human_desc': tool.metadata.description.en_US,
'parameters': tool.spec['parameters'],
'source': 'plugin',
'source_name': tool.owner,
'source_id': tool.owner,
}
)
return catalog
async def has_tool(self, name: str) -> bool:
"""检查工具是否存在"""
for tool in await self.ap.plugin_connector.list_tools():
src/langbot/pkg/provider/tools/toolmgr.py
+44 -1
View File
@@ -58,6 +58,8 @@ class ToolManager:
self,
bound_plugins: list[str] | None = None,
bound_mcp_servers: list[str] | None = None,
include_skill_authoring: bool = False,
include_mcp_resource_tools: bool = True,
) -> list[resource_tool.LLMTool]:
all_functions: list[resource_tool.LLMTool] = []
@@ -68,10 +70,51 @@ class ToolManager:
# capability-gated surface.
all_functions.extend(await self.skill_tool_loader.get_tools())
all_functions.extend(await self.plugin_tool_loader.get_tools(bound_plugins))
all_functions.extend(await self.mcp_tool_loader.get_tools(bound_mcp_servers))
all_functions.extend(
await self.mcp_tool_loader.get_tools(
bound_mcp_servers,
include_resource_tools=include_mcp_resource_tools,
)
)
return all_functions
async def get_tool_catalog(
self,
bound_plugins: list[str] | None = None,
bound_mcp_servers: list[str] | None = None,
include_skill_authoring: bool = False,
include_mcp_resource_tools: bool = False,
) -> list[dict[str, typing.Any]]:
catalog: list[dict[str, typing.Any]] = []
def append_tools(source: str, source_name: str, tools: list[resource_tool.LLMTool]) -> None:
for tool in tools:
catalog.append(
{
'name': tool.name,
'description': tool.description,
'human_desc': tool.human_desc,
'parameters': tool.parameters,
'source': source,
'source_name': source_name,
}
)
append_tools('builtin', 'LangBot', await self.native_tool_loader.get_tools())
if include_skill_authoring:
append_tools('skill', 'LangBot', await self.skill_tool_loader.get_tools())
catalog.extend(await self.plugin_tool_loader.get_tool_catalog(bound_plugins))
if self.mcp_tool_loader:
for item in await self.mcp_tool_loader.get_tool_catalog(
bound_mcp_servers,
include_resource_tools=include_mcp_resource_tools,
):
catalog.append(item)
return catalog
async def get_tool_by_name(self, name: str) -> tool_loader.ToolLookupResult | None:
"""Get tool by name from any active loader."""
for active_loader in (
src/langbot/pkg/survey/manager.py
+43 -3
View File
@@ -159,6 +159,21 @@ class SurveyManager:
"""Clear the pending survey (after user responds or dismisses)."""
self._pending_survey = None
async def _build_base_metadata(self, user_email: str | None = None) -> dict:
metadata = {
'version': constants.semantic_version,
'instance_id': constants.instance_id,
}
if user_email:
metadata['login_account'] = user_email
try:
user_obj = await self.ap.user_service.get_user_by_email(user_email)
metadata['account_type'] = getattr(user_obj, 'account_type', '') or 'local'
metadata['space_account_uuid'] = getattr(user_obj, 'space_account_uuid', '') or ''
except Exception:
pass
return metadata
async def submit_response(self, survey_id: str, answers: dict, completed: bool = True) -> bool:
"""Submit a survey response to Space."""
if not self._is_space_configured():
@@ -169,9 +184,7 @@ class SurveyManager:
'survey_id': survey_id,
'instance_id': constants.instance_id,
'answers': answers,
'metadata': {
'version': constants.semantic_version,
},
'metadata': await self._build_base_metadata(),
'completed': completed,
}
async with httpx.AsyncClient(timeout=httpx.Timeout(10)) as client:
@@ -183,6 +196,33 @@ class SurveyManager:
self.ap.logger.warning(f'Failed to submit survey response: {e}')
return False
async def submit_feedback(
self,
content: str,
attachments: list[dict],
user_email: str | None = None,
) -> bool:
"""Submit an on-demand user feedback item to Space."""
if not self._is_space_configured():
return False
try:
url = f'{self._space_url}/api/v1/survey/feedback'
metadata = await self._build_base_metadata(user_email)
payload = {
'instance_id': constants.instance_id,
'content': content,
'attachments': attachments,
'metadata': metadata,
}
async with httpx.AsyncClient(timeout=httpx.Timeout(30)) as client:
resp = await client.post(url, json=payload)
if resp.status_code == 200:
return True
self.ap.logger.warning(f'Failed to submit feedback: {resp.status_code} {resp.text[:200]}')
except Exception as e:
self.ap.logger.warning(f'Failed to submit feedback: {e}')
return False
async def dismiss_survey(self, survey_id: str) -> bool:
"""Dismiss a survey."""
if not self._is_space_configured():
src/langbot/templates/config.yaml
+2 -1
View File
@@ -1,4 +1,3 @@
admins: []
api:
port: 5300
webhook_prefix: 'http://127.0.0.1:5300'
@@ -157,6 +156,8 @@ box:
- './data/box'
- '/tmp'
workspace_quota_mb: null # Optional disk quota override (>= 0). null = profile default.
docker:
cpu_limit_enabled: true # When false, Docker sandbox containers are started without --cpus. Memory and PID limits still apply.
e2b:
api_key: '' # Can also be set via E2B_API_KEY env var.
api_url: '' # Custom API URL for self-hosted deployments.
src/langbot/templates/metadata/pipeline/ai.yaml
+1 -1
View File
@@ -40,4 +40,4 @@ stages:
default: 0
# Runner config stages are dynamically added from AgentRunnerRegistry
# Each plugin runner's config schema is added as a separate stage
# The stage name matches the runner id for frontend matching
# The stage name matches the runner id for frontend matching