mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-07-22 04:16:07 +00:00
huanghuoguoguo
5f65f962c5
Expose skill tools (activate/register_skill/native exec) like native tools instead of gating them behind the skill_authoring capability: - toolmgr.get_all_tools drops include_skill_authoring; SkillToolLoader self-gates on sandbox + skill_mgr - preproc drops the include_skill_authoring branch; pipeline-bound skills and the skills resource gate on skill_mgr presence Persist activated skills into host.activated_skills conversation state so they survive across runs (host writes at activate; last-write-wins); drop the dead restore_activated_skills helper. Prefill ToolResource.parameters host-side (tool_mgr.get_tool_schema) so runners build LLM tools without per-tool get_tool_detail round-trips. Align agent-runner-pluginization design docs to the all-tool model.
64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import quart
|
|
|
|
from ... import group
|
|
|
|
|
|
@group.group_class('tools', '/api/v1/tools')
|
|
class ToolsRouterGroup(group.RouterGroup):
|
|
async def initialize(self) -> None:
|
|
@self.route('', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
|
|
async def _() -> str:
|
|
"""获取所有可用工具列表"""
|
|
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
|
|
|
|
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')
|
|
|
|
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()
|
|
|
|
for tool in tools:
|
|
if tool.name == tool_name:
|
|
return self.success(
|
|
data={
|
|
'tool': {
|
|
'name': tool.name,
|
|
'description': tool.description,
|
|
'human_desc': tool.human_desc,
|
|
'parameters': tool.parameters,
|
|
}
|
|
}
|
|
)
|
|
|
|
return self.http_status(404, -1, f'Tool not found: {tool_name}')
|