chore(agent-runner): merge split tool runtime base

# Conflicts:
#	src/langbot/pkg/box/workspace.py
#	src/langbot/pkg/provider/tools/loaders/mcp_stdio.py
#	src/langbot/pkg/provider/tools/loaders/native.py
#	src/langbot/pkg/provider/tools/loaders/skill.py
#	tests/unit_tests/box/test_workspace.py
#	tests/unit_tests/provider/test_mcp_box_integration.py
This commit is contained in:
huanghuoguoguo
2026-06-14 21:22:05 +08:00
28 changed files with 548 additions and 53 deletions
@@ -271,6 +271,20 @@ class PluginsRouterGroup(group.RouterGroup):
readme = await self.ap.plugin_connector.get_plugin_readme(author, plugin_name, language=language)
return self.success(data={'readme': readme})
@self.route(
'/<author>/<plugin_name>/logs',
methods=['GET'],
auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
)
async def _(author: str, plugin_name: str) -> quart.Response:
try:
limit = int(quart.request.args.get('limit', 200))
except (TypeError, ValueError):
limit = 200
level = quart.request.args.get('level') or None
logs = await self.ap.plugin_connector.get_plugin_logs(author, plugin_name, limit=limit, level=level)
return self.success(data={'logs': logs})
@self.route(
'/<author>/<plugin_name>/icon',
methods=['GET'],
@@ -202,6 +202,16 @@ class LoadConfigStage(stage.BootingStage):
constants.instance_id = new_id
constants.edition = ap.instance_config.data.get('system', {}).get('edition', 'community')
# Instance creation timestamp: sourced from data/labels/instance_id.json.
# Instances created before this field existed (or supplied via
# system.instance_id) won't have it, so backfill with the current time
# and persist it via the dump below — from then on it stays stable.
instance_create_ts = ap.instance_id.data.get('instance_create_ts', 0)
if not isinstance(instance_create_ts, int) or instance_create_ts <= 0:
instance_create_ts = int(time.time())
ap.instance_id.data['instance_create_ts'] = instance_create_ts
constants.instance_create_ts = instance_create_ts
print(f'LangBot instance id: {constants.instance_id}')
print(f'LangBot edition: {constants.edition}')
@@ -84,6 +84,18 @@ class WebPageBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
):
self.listeners.pop(event_type, None)
async def is_stream_output_supported(self) -> bool:
"""Delegate stream output check to ws_adapter."""
if self._ws_adapter is not None:
return await self._ws_adapter.is_stream_output_supported()
return False
async def create_message_card(self, message_id: str | int, event: platform_events.MessageEvent) -> bool:
"""Delegate create_message_card to ws_adapter."""
if self._ws_adapter is not None:
return await self._ws_adapter.create_message_card(message_id, event)
return False
async def is_muted(self, group_id: int) -> bool:
return False
+10
View File
@@ -708,6 +708,16 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
async def get_plugin_readme(self, plugin_author: str, plugin_name: str, language: str = 'en') -> str:
return await self.handler.get_plugin_readme(plugin_author, plugin_name, language)
async def get_plugin_logs(
self,
plugin_author: str,
plugin_name: str,
limit: int = 200,
level: str | None = None,
) -> list[dict[str, Any]]:
# Not cached: logs are live and change constantly.
return await self.handler.get_plugin_logs(plugin_author, plugin_name, limit, level)
@alru_cache(ttl=5 * 60)
async def get_plugin_assets(self, plugin_author: str, plugin_name: str, filepath: str) -> dict[str, Any]:
return await self.handler.get_plugin_assets(plugin_author, plugin_name, filepath)
+25
View File
@@ -2515,6 +2515,31 @@ class RuntimeConnectionHandler(handler.Handler):
return readme_bytes.decode('utf-8')
async def get_plugin_logs(
self,
plugin_author: str,
plugin_name: str,
limit: int = 200,
level: str | None = None,
) -> list[dict[str, Any]]:
"""Get recent log lines captured from the plugin's stderr."""
try:
result = await self.call_action(
LangBotToRuntimeAction.GET_PLUGIN_LOGS,
{
'plugin_author': plugin_author,
'plugin_name': plugin_name,
'limit': limit,
'level': level,
},
timeout=20,
)
except Exception:
traceback.print_exc()
return []
return result.get('logs', [])
async def get_plugin_assets(self, plugin_author: str, plugin_name: str, filepath: str) -> dict[str, Any]:
"""Get plugin assets"""
result = await self.call_action(
+18 -3
View File
@@ -12,6 +12,19 @@ import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
import langbot_plugin.api.entities.builtin.provider.message as provider_message
LLM_USAGE_QUERY_VARIABLE = '_llm_usage'
STREAM_USAGE_QUERY_VARIABLE = '_stream_usage'
def _store_llm_usage(query: pipeline_query.Query | None, usage_info: dict | None) -> None:
"""Store the latest provider usage on the query for upstream action handlers."""
if query is None or not usage_info:
return
if query.variables is None:
query.variables = {}
query.variables[LLM_USAGE_QUERY_VARIABLE] = dict(usage_info)
class RuntimeProvider:
"""运行时模型提供商"""
@@ -67,6 +80,7 @@ class RuntimeProvider:
if isinstance(result, tuple):
msg, usage_info = result
if usage_info:
_store_llm_usage(query, usage_info)
input_tokens = usage_info.get('prompt_tokens', 0)
output_tokens = usage_info.get('completion_tokens', 0)
return msg
@@ -146,11 +160,12 @@ class RuntimeProvider:
if query:
if query.variables is None:
query.variables = {}
if '_stream_usage' in query.variables:
usage_info = query.variables['_stream_usage']
if STREAM_USAGE_QUERY_VARIABLE in query.variables:
usage_info = query.variables[STREAM_USAGE_QUERY_VARIABLE]
_store_llm_usage(query, usage_info)
input_tokens = usage_info.get('prompt_tokens', 0)
output_tokens = usage_info.get('completion_tokens', 0)
del query.variables['_stream_usage']
del query.variables[STREAM_USAGE_QUERY_VARIABLE]
except Exception as e:
status = 'error'
error_message = str(e)
@@ -75,22 +75,33 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
continue
return False
@staticmethod
def _positive_int(value: typing.Any) -> int | None:
if isinstance(value, bool):
return None
if isinstance(value, int) and value > 0:
return value
if isinstance(value, str) and value.isdigit():
parsed_value = int(value)
if parsed_value > 0:
return parsed_value
return None
def _context_length_from_scan_payload(self, model_payload: dict[str, typing.Any] | None) -> int | None:
if not model_payload:
return None
for field_name in ('context_length', 'context_window', 'max_context_length'):
value = model_payload.get(field_name)
if isinstance(value, bool):
continue
if isinstance(value, int) and value > 0:
return value
if isinstance(value, str) and value.isdigit():
parsed_value = int(value)
if parsed_value > 0:
return parsed_value
context_length = self._positive_int(model_payload.get(field_name))
if context_length is not None:
return context_length
return None
def _context_length_from_litellm_model_info(self, model_info: typing.Any) -> int | None:
if isinstance(model_info, dict):
return self._positive_int(model_info.get('max_input_tokens'))
return self._positive_int(getattr(model_info, 'max_input_tokens', None))
def _metadata_provider_candidates(self, model_name: str) -> list[str]:
normalized_model_name = (model_name or '').lower()
candidates = []
@@ -126,7 +137,7 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
return None
def _safe_context_length(self, model_name: str) -> int | None:
helper = getattr(litellm, 'get_max_tokens', None)
helper = getattr(litellm, 'get_model_info', None)
if not callable(helper):
return self._known_context_length_fallback(model_name)
@@ -143,11 +154,12 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
continue
tried_candidates.append(candidate)
try:
max_tokens = helper(candidate)
model_info = helper(candidate)
except Exception:
continue
if isinstance(max_tokens, int) and max_tokens > 0:
return max_tokens
context_length = self._context_length_from_litellm_model_info(model_info)
if context_length is not None:
return context_length
return self._known_context_length_fallback(model_name)
def _supports_function_calling(self, model_name: str) -> bool:
@@ -250,32 +262,82 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
- dict with the same keys
- missing ``total_tokens`` (derived from prompt + completion)
- ``None`` / partially-populated usage (defaults to 0)
- provider-specific token details, including cache token counters
"""
if usage is None:
return {'prompt_tokens': 0, 'completion_tokens': 0, 'total_tokens': 0}
def _get(key: str) -> typing.Any:
if isinstance(usage, dict):
return usage.get(key)
return getattr(usage, key, None)
def _plain_value(value: typing.Any) -> typing.Any:
if value is None:
return None
if isinstance(value, dict):
return {k: _plain_value(v) for k, v in value.items() if v is not None}
if isinstance(value, (list, tuple)):
return [_plain_value(v) for v in value]
prompt_tokens = _get('prompt_tokens') or 0
completion_tokens = _get('completion_tokens') or 0
total_tokens = _get('total_tokens') or 0
model_dump = getattr(value, 'model_dump', None)
if callable(model_dump):
try:
dumped = model_dump()
if isinstance(dumped, dict):
return _plain_value(dumped)
except Exception:
pass
return value
def _usage_dict(value: typing.Any) -> dict[str, typing.Any]:
if value is None:
return {}
plain = _plain_value(value)
if isinstance(plain, dict):
return plain
def _is_mock_attr(attr: typing.Any) -> bool:
return type(attr).__module__.startswith('unittest.mock')
data: dict[str, typing.Any] = {}
for key in (
'prompt_tokens',
'completion_tokens',
'total_tokens',
'prompt_tokens_details',
'completion_tokens_details',
'cache_creation_input_tokens',
'cache_read_input_tokens',
'input_token_details',
'output_token_details',
):
attr_value = getattr(value, key, None)
if attr_value is not None and not _is_mock_attr(attr_value):
data[key] = _plain_value(attr_value)
return data
def _to_int(value: typing.Any) -> int:
try:
return int(value or 0)
except (TypeError, ValueError):
return 0
normalized = _usage_dict(usage)
prompt_tokens = _to_int(normalized.get('prompt_tokens'))
completion_tokens = _to_int(normalized.get('completion_tokens'))
total_tokens = _to_int(normalized.get('total_tokens'))
# Some providers omit total_tokens in streaming usage; derive it.
if not total_tokens:
total_tokens = prompt_tokens + completion_tokens
return {
'prompt_tokens': int(prompt_tokens),
'completion_tokens': int(completion_tokens),
'total_tokens': int(total_tokens),
}
normalized['prompt_tokens'] = prompt_tokens
normalized['completion_tokens'] = completion_tokens
normalized['total_tokens'] = total_tokens
return normalized
def _extract_usage(self, response) -> dict:
def _extract_usage(self, response) -> dict | None:
"""Extract usage info from a non-streaming LiteLLM response."""
return self._normalize_usage(getattr(response, 'usage', None))
usage = getattr(response, 'usage', None)
if usage is None:
return None
return self._normalize_usage(usage)
@staticmethod
def _as_dict(value: typing.Any) -> dict:
@@ -474,7 +536,7 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
if query is not None:
if query.variables is None:
query.variables = {}
query.variables['_stream_usage'] = usage_info
query.variables[requester.STREAM_USAGE_QUERY_VARIABLE] = usage_info
if not hasattr(chunk, 'choices') or not chunk.choices:
continue
@@ -117,6 +117,7 @@ class SkillToolLoader(loader.ToolLoader):
'activated': True,
'skill_name': skill_name,
'mount_path': mount_path,
'activated_skill_names': skill_loader.get_activated_skill_names(query),
'content': result_content,
}
+1
View File
@@ -109,6 +109,7 @@ async def build_heartbeat_payload(ap: core_app.Application) -> dict:
'query_id': '',
'version': constants.semantic_version,
'instance_id': constants.instance_id,
'instance_create_ts': constants.instance_create_ts,
'edition': constants.edition,
'features': features,
'timestamp': datetime.now(timezone.utc).isoformat(),
+8
View File
@@ -16,3 +16,11 @@ debug_mode = False
edition = 'community'
instance_id = ''
instance_create_ts = 0
"""Unix timestamp (seconds) of when this instance was first created.
Sourced from ``data/labels/instance_id.json``. Backfilled to the current
time for instances created before this field existed, so it is always a
positive value once load_config has run.
"""