diff --git a/src/langbot/pkg/agent/runner/interaction_manager.py b/src/langbot/pkg/agent/runner/interaction_manager.py index efa09bb5d..54b024283 100644 --- a/src/langbot/pkg/agent/runner/interaction_manager.py +++ b/src/langbot/pkg/agent/runner/interaction_manager.py @@ -207,6 +207,122 @@ class InteractionManager: raise return True + async def present_tool_request( + self, + *, + request_payload: dict[str, typing.Any], + session: typing.Mapping[str, typing.Any], + adapter: typing.Any, + ) -> dict[str, typing.Any]: + """Persist and deliver an Agent-authored semantic card using Host-owned authority.""" + authorization = session.get('authorization') if isinstance(session, typing.Mapping) else None + if not isinstance(authorization, dict): + raise ValueError('Interactive card run authorization is unavailable') + if authorization.get('processor_type') != 'agent': + raise ValueError('Interactive card tools are currently available only to independent Agents') + if authorization.get('interactions_enabled') is not True: + raise ValueError('Interactive cards are disabled for this Agent binding') + + run_id = str(session.get('run_id') or '').strip() + runner_id = str(session.get('runner_id') or '').strip() + binding_id = str(authorization.get('binding_id') or '').strip() + processor_id = str(authorization.get('processor_id') or '').strip() + if not all((run_id, runner_id, binding_id, processor_id)): + raise ValueError('Interactive card run identity is incomplete') + try: + request = HostInteractionRequest.model_validate(request_payload) + except pydantic.ValidationError as exc: + raise ValueError(f'Invalid interactive card: {exc}') from exc + self._validate_request_identity(request, runner_id) + + request_data = request.model_dump(mode='json') + now = int(time.time()) + expires_at = request.expires_at or now + DEFAULT_INTERACTION_TTL_SECONDS + if expires_at <= now or expires_at > now + MAX_INTERACTION_TTL_SECONDS: + raise ValueError('Interactive card expiry must be in the future and within 24 hours') + request_data['expires_at'] = expires_at + if len(json.dumps(request_data, ensure_ascii=False).encode('utf-8')) > INTERACTION_PAYLOAD_MAX_BYTES: + raise ValueError('Interactive card exceeds 256 KiB') + + platform_context = authorization.get('platform_context') + if not isinstance(platform_context, dict): + raise ValueError('Interactive card platform context is unavailable') + delivery = platform_context.get('delivery') if isinstance(platform_context.get('delivery'), dict) else {} + self._validate_tool_delivery_capabilities(request, delivery) + reply_target = delivery.get('reply_target') if isinstance(delivery.get('reply_target'), dict) else {} + target_type = reply_target.get('target_type') + target_id = reply_target.get('target_id') + if not target_type or not target_id: + raise ValueError('Interactive card has no authorized reply target') + callback_conversation_id = f'{target_type}_{target_id}' + actor = platform_context.get('actor') if isinstance(platform_context.get('actor'), dict) else {} + + _, callback_token = await self.store.create_request( + interaction_id=request.interaction_id, + run_id=run_id, + binding_id=binding_id, + runner_id=runner_id, + processor_type='agent', + processor_id=processor_id, + request=request_data, + delivery_target=reply_target, + bot_id=authorization.get('bot_id'), + workspace_id=authorization.get('workspace_id'), + conversation_id=callback_conversation_id, + thread_id=authorization.get('thread_id'), + actor_id=actor.get('actor_id'), + expires_at=expires_at, + ) + try: + if not self._supports_interaction_delivery(adapter): + raise ValueError('The current platform does not support interactive cards') + delivery_result = await adapter.call_platform_api( + INTERACTION_DELIVERY_API, + { + 'request': request_data, + 'callback_token': callback_token, + 'reply_target': reply_target, + }, + ) + if isinstance(delivery_result, dict): + await self.store.record_delivery_success(run_id, request.interaction_id, delivery_result) + except Exception as exc: + await self.store.mark_delivery_failed(run_id, request.interaction_id, str(exc)) + raise + return { + 'ok': True, + 'status': 'pending', + 'interaction_id': request.interaction_id, + 'expires_at': expires_at, + } + + @staticmethod + def _validate_tool_delivery_capabilities( + request: HostInteractionRequest, + delivery: dict[str, typing.Any], + ) -> None: + capabilities = delivery.get('interactions') + if not isinstance(capabilities, dict): + raise ValueError('The current platform did not declare interactive card capabilities') + if request.fields and request.actions: + raise ValueError('An interactive card step cannot combine fields and actions') + field_types = set(capabilities.get('field_types') or []) + unsupported_types = {field.type for field in request.fields} - field_types + if unsupported_types: + raise ValueError( + 'The current platform does not support interaction fields: ' + ', '.join(sorted(unsupported_types)) + ) + max_fields = capabilities.get('max_fields') + if isinstance(max_fields, int) and len(request.fields) > max_fields: + raise ValueError(f'The current platform supports at most {max_fields} interaction fields') + action_styles = set(capabilities.get('action_styles') or []) + unsupported_styles = {action.style for action in request.actions} - action_styles + if unsupported_styles: + raise ValueError( + 'The current platform does not support interaction action styles: ' + + ', '.join(sorted(unsupported_styles)) + ) + @staticmethod def _pipeline_admission(binding, descriptor, adapter_context): if binding.processor_type != 'pipeline': diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index 5818902eb..b88d0eba9 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -297,6 +297,10 @@ class AgentRunOrchestrator: execution_query=execution_query, platform_context=freeze_platform_context(event), reply_streams=reply_streams, + binding_id=binding.binding_id, + processor_type=binding.processor_type, + processor_id=binding.processor_id or binding.agent_id, + interactions_enabled=binding.delivery_policy.enable_interactions, ) event_log_id = await self.journal.write_event_log( diff --git a/src/langbot/pkg/agent/runner/platform_tools.py b/src/langbot/pkg/agent/runner/platform_tools.py index 5b7dc6509..cf65feaf3 100644 --- a/src/langbot/pkg/agent/runner/platform_tools.py +++ b/src/langbot/pkg/agent/runner/platform_tools.py @@ -4,7 +4,9 @@ from __future__ import annotations import copy import fnmatch +import time import typing +import uuid from dataclasses import dataclass import langbot_plugin.api.entities.builtin.platform.message as platform_message @@ -44,8 +46,77 @@ _TARGET_TYPE = {'type': 'string', 'enum': ['person', 'group']} _CHAT_TYPE = {'type': 'string', 'enum': ['person', 'private', 'group']} _APPROVE = {'type': 'boolean', 'description': 'true to accept; false to reject'} +PRESENT_INTERACTIVE_CARD_TOOL = 'present_interactive_card' + +_INTERACTION_OPTION_SCHEMA = { + 'type': 'object', + 'properties': { + 'value': {'type': 'string', 'minLength': 1, 'maxLength': 512}, + 'label': {'type': 'string', 'minLength': 1, 'maxLength': 512}, + 'description': {'type': 'string', 'maxLength': 2000}, + }, + 'required': ['value', 'label'], + 'additionalProperties': False, +} +_INTERACTION_FIELD_SCHEMA = { + 'type': 'object', + 'properties': { + 'id': {'type': 'string', 'minLength': 1, 'maxLength': 128}, + 'label': {'type': 'string', 'minLength': 1, 'maxLength': 512}, + 'type': { + 'type': 'string', + 'enum': ['text', 'textarea', 'select', 'multiselect', 'number', 'boolean', 'file'], + }, + 'required': {'type': 'boolean'}, + 'options': {'type': 'array', 'items': _INTERACTION_OPTION_SCHEMA, 'maxItems': 100}, + 'placeholder': {'type': 'string', 'maxLength': 2000}, + 'default': {}, + }, + 'required': ['id', 'label', 'type'], + 'additionalProperties': False, +} +_INTERACTION_ACTION_SCHEMA = { + 'type': 'object', + 'properties': { + 'id': {'type': 'string', 'minLength': 1, 'maxLength': 128}, + 'label': {'type': 'string', 'minLength': 1, 'maxLength': 512}, + 'style': {'type': 'string', 'enum': ['default', 'primary', 'danger']}, + }, + 'required': ['id', 'label'], + 'additionalProperties': False, +} + PLATFORM_TOOL_DEFINITIONS: tuple[PlatformToolDefinition, ...] = ( + PlatformToolDefinition( + PRESENT_INTERACTIVE_CARD_TOOL, + 'interaction.request', + 'event', + 'interaction', + 'write', + {'en_US': 'Present interactive card'}, + { + 'en_US': ( + 'Present a native interactive card in the current conversation. Define semantic buttons, selects, ' + 'and input fields; LangBot owns the target, callback token, validation, and platform rendering. ' + 'Each card is one step: provide either actions or fields, never both. ' + 'The user submission will start a new interaction.submitted turn for this Agent.' + ), + }, + _object_schema( + { + 'kind': {'type': 'string', 'enum': ['form', 'confirmation', 'choice']}, + 'title': {'type': 'string', 'minLength': 1, 'maxLength': 1000}, + 'description': {'type': 'string', 'maxLength': 10000}, + 'fields': {'type': 'array', 'items': _INTERACTION_FIELD_SCHEMA, 'maxItems': 50}, + 'actions': {'type': 'array', 'items': _INTERACTION_ACTION_SCHEMA, 'maxItems': 20}, + 'expires_in_seconds': {'type': 'integer', 'minimum': 60, 'maximum': 86400}, + }, + ['title'], + ), + ('message.*', 'interaction.submitted'), + 'reply_target', + ), PlatformToolDefinition( 'event_reply', 'send_message', @@ -479,13 +550,26 @@ def build_platform_tool_resources( unavailable.append({'name': name, 'reason': reason}) continue assert definition is not None + parameters = copy.deepcopy(definition.parameters) + if definition.name == PRESENT_INTERACTIVE_CARD_TOOL: + capabilities = getattr(event.delivery, 'interactions', None) + field_types = list(getattr(capabilities, 'field_types', []) or []) + action_styles = list(getattr(capabilities, 'action_styles', []) or []) + max_fields = getattr(capabilities, 'max_fields', None) + field_schema = parameters['properties']['fields'] + if field_types: + field_schema['items']['properties']['type']['enum'] = field_types + if isinstance(max_fields, int): + field_schema['maxItems'] = max_fields + if action_styles: + parameters['properties']['actions']['items']['properties']['style']['enum'] = action_styles resources.append( { 'tool_name': definition.name, 'tool_type': 'platform', 'description': definition.description.get('en_US') or definition.description.get('zh_Hans'), 'operations': list(operations), - 'parameters': copy.deepcopy(definition.parameters), + 'parameters': parameters, 'source': 'platform', 'source_id': definition.name, } @@ -537,6 +621,8 @@ def _event_params( subject.get('subject_id') if subject.get('subject_type') == 'group' else None ) actor_id = actor.get('actor_id') + if definition.name == PRESENT_INTERACTIVE_CARD_TOOL: + return dict(parameters) if definition.binding == 'reply_target': return { 'target_type': reply_target.get('target_type'), @@ -608,6 +694,49 @@ def _normalize_platform_params( return dict(parameters) +def _interactive_card_request(parameters: dict[str, typing.Any]) -> dict[str, typing.Any]: + fields = copy.deepcopy(parameters.get('fields') or []) + actions = copy.deepcopy(parameters.get('actions') or []) + if not isinstance(fields, list) or not isinstance(actions, list): + raise ValueError('fields and actions must be arrays') + if not fields and not actions: + raise ValueError('An interactive card requires at least one field or action') + if fields and actions: + raise ValueError('An interactive card step cannot combine fields and actions') + + title = _require_string(parameters, 'title') + description = parameters.get('description') + lines = [title] + if isinstance(description, str) and description.strip(): + lines.extend(['', description.strip()]) + for field in fields: + if not isinstance(field, dict): + raise ValueError('Every interaction field must be an object') + label = str(field.get('label') or field.get('id') or 'Input') + options = field.get('options') if isinstance(field.get('options'), list) else [] + option_labels = [str(item.get('label')) for item in options if isinstance(item, dict) and item.get('label')] + lines.append(f'{label}: {", ".join(option_labels) if option_labels else "input required"}') + if actions: + lines.append( + 'Actions: ' + + ', '.join(str(item.get('label')) for item in actions if isinstance(item, dict) and item.get('label')) + ) + + expires_in = parameters.get('expires_in_seconds', 30 * 60) + if isinstance(expires_in, bool) or not isinstance(expires_in, int) or not 60 <= expires_in <= 86400: + raise ValueError('expires_in_seconds must be an integer from 60 to 86400') + return { + 'interaction_id': f'card-{uuid.uuid4().hex}', + 'kind': parameters.get('kind', 'form'), + 'title': title, + 'description': description, + 'fields': fields, + 'actions': actions, + 'expires_at': int(time.time()) + expires_in, + 'fallback_text': '\n'.join(lines), + } + + def resolve_platform_api_call(session, bot_uuid, action, params, context_tool=None): """Resolve an adapter call against Host-frozen tool grants and event targets.""" authorization = session.get('authorization') or {} @@ -740,6 +869,15 @@ async def execute_platform_tool( ) if definition.api not in supported_apis and not reply_via_source: raise ValueError(f'Platform API {definition.api} is no longer supported by bot {bot_id}') + if definition.name == PRESENT_INTERACTIVE_CARD_TOOL: + manager = getattr(getattr(ap, 'agent_run_orchestrator', None), 'interaction_manager', None) + if manager is None: + raise ValueError('Interactive card delivery is unavailable') + return await manager.present_tool_request( + request_payload=_interactive_card_request(normalized), + session=session, + adapter=bot.adapter, + ) api_func = getattr(bot.adapter, definition.api, None) if reply_via_source: @@ -877,6 +1015,7 @@ def _execute_mock_platform_tool( __all__ = [ 'PLATFORM_TOOL_DEFINITIONS', + 'PRESENT_INTERACTIVE_CARD_TOOL', 'build_platform_tool_resources', 'execute_platform_tool', 'freeze_platform_context', diff --git a/src/langbot/pkg/agent/runner/resource_builder.py b/src/langbot/pkg/agent/runner/resource_builder.py index 76d5dd1c4..19cfca3e5 100644 --- a/src/langbot/pkg/agent/runner/resource_builder.py +++ b/src/langbot/pkg/agent/runner/resource_builder.py @@ -20,7 +20,7 @@ from .host_models import AgentEventEnvelope, AgentBinding from .resource_policy import ResourcePolicyProjector from ...provider.tools.loaders.mcp import MCP_TOOL_LIST_RESOURCES, MCP_TOOL_READ_RESOURCE from ...provider.tools.toolmgr import ToolSourceRef -from .platform_tools import build_platform_tool_resources +from .platform_tools import PRESENT_INTERACTIVE_CARD_TOOL, build_platform_tool_resources class AgentResourceBuilder: @@ -88,9 +88,14 @@ class AgentResourceBuilder: runner_config, ) runner_uses_host_tools = config_schema.uses_host_tools(descriptor) + selected_platform_tools = list(resource_policy.allowed_platform_tool_names or []) + if binding.processor_type != 'agent' or not binding.delivery_policy.enable_interactions: + selected_platform_tools = [ + name for name in selected_platform_tools if name != PRESENT_INTERACTIVE_CARD_TOOL + ] platform_tools, platform_capabilities = build_platform_tool_resources( event, - resource_policy.allowed_platform_tool_names, + selected_platform_tools, ( [operation for operation in ('detail', 'call') if operation in set(manifest_perms.tools)] if runner_uses_host_tools diff --git a/src/langbot/pkg/agent/runner/session_registry.py b/src/langbot/pkg/agent/runner/session_registry.py index e7c12716b..1af30d04a 100644 --- a/src/langbot/pkg/agent/runner/session_registry.py +++ b/src/langbot/pkg/agent/runner/session_registry.py @@ -50,6 +50,10 @@ class RunAuthorizationSnapshot(typing.TypedDict): platform_context: dict[str, typing.Any] authorized_ids: dict[str, set[str]] authorized_operations: dict[str, dict[str, set[str]]] + binding_id: str | None + processor_type: str | None + processor_id: str | None + interactions_enabled: bool SteeringQueueItem = dict[str, typing.Any] @@ -117,6 +121,10 @@ class AgentRunSessionRegistry: execution_query: pipeline_query.Query | None = None, platform_context: dict[str, typing.Any] | None = None, reply_streams: typing.Any = None, + binding_id: str | None = None, + processor_type: str | None = None, + processor_id: str | None = None, + interactions_enabled: bool = False, ) -> None: """Register a new agent run session. @@ -162,6 +170,10 @@ class AgentRunSessionRegistry: 'platform_context': copy.deepcopy(platform_context or {}), 'authorized_ids': self._build_authorized_ids(resources_snapshot), 'authorized_operations': self._build_authorized_operations(resources_snapshot), + 'binding_id': binding_id, + 'processor_type': processor_type, + 'processor_id': processor_id, + 'interactions_enabled': interactions_enabled, } session: AgentRunSession = { diff --git a/tests/unit_tests/agent/test_interaction_manager.py b/tests/unit_tests/agent/test_interaction_manager.py index 821c24fc7..06eb8e516 100644 --- a/tests/unit_tests/agent/test_interaction_manager.py +++ b/tests/unit_tests/agent/test_interaction_manager.py @@ -190,6 +190,66 @@ async def test_structured_delivery_uses_frozen_target_and_persists_request(store assert record['delivery_result']['message_id'] == 'message-1' +@pytest.mark.asyncio +async def test_agent_tool_card_uses_host_token_and_persists_resume_target(store, event): + adapter = FakeAdapter(supports_interactions=True) + manager = InteractionManager(SimpleNamespace(), store=store) + event.delivery.interactions = InteractionDeliveryCapabilities( + field_types=['text', 'textarea', 'select', 'number'], + action_styles=['default', 'primary', 'danger'], + supports_updates=True, + max_fields=1, + ) + session = { + 'run_id': 'agent-run-1', + 'runner_id': 'plugin:test/CardAgent/default', + 'authorization': { + 'binding_id': 'agent-binding-1', + 'processor_type': 'agent', + 'processor_id': 'agent-1', + 'interactions_enabled': True, + 'bot_id': 'bot-1', + 'workspace_id': 'workspace-1', + 'thread_id': None, + 'platform_context': { + 'actor': {'actor_type': 'user', 'actor_id': 'user-1'}, + 'delivery': event.delivery.model_dump(mode='json'), + }, + }, + } + request = { + 'interaction_id': 'card-host-generated', + 'kind': 'choice', + 'title': 'Choose priority', + 'fields': [ + { + 'id': 'priority', + 'label': 'Priority', + 'type': 'select', + 'options': [{'value': 'high', 'label': 'High'}], + } + ], + 'actions': [], + 'expires_at': int(time.time()) + 600, + 'fallback_text': 'Choose priority', + } + + result = await manager.present_tool_request(request_payload=request, session=session, adapter=adapter) + + assert result['status'] == 'pending' + assert 'callback_token' not in result + params = adapter.actions[0][1] + assert params['callback_token'] + assert params['reply_target'] == {'target_type': 'group', 'target_id': 'chat-1'} + record = await store.get_request('agent-run-1', 'card-host-generated') + assert record is not None + assert record['processor_type'] == 'agent' + assert record['processor_id'] == 'agent-1' + assert record['binding_id'] == 'agent-binding-1' + assert record['conversation_id'] == 'group_chat-1' + assert record['actor_id'] == 'user-1' + + @pytest.mark.asyncio async def test_continuous_interaction_reuses_submitted_platform_presentation(store, event, binding): adapter = FakeAdapter(supports_interactions=True) diff --git a/tests/unit_tests/agent/test_platform_tools.py b/tests/unit_tests/agent/test_platform_tools.py index a4dbf3190..73607aa1c 100644 --- a/tests/unit_tests/agent/test_platform_tools.py +++ b/tests/unit_tests/agent/test_platform_tools.py @@ -6,6 +6,7 @@ from langbot_plugin.api.entities.builtin.runner import ( ActorContext, AgentInput, DeliveryContext, + InteractionDeliveryCapabilities, RawEventRef, SubjectContext, ) @@ -16,6 +17,7 @@ from langbot_plugin.api.entities.builtin.platform import events as platform_even from langbot.pkg.agent.runner.host_models import AgentEventEnvelope from langbot.pkg.agent.runner.platform_tools import ( PLATFORM_TOOL_DEFINITIONS, + PRESENT_INTERACTIVE_CARD_TOOL, build_platform_tool_resources, execute_platform_tool, freeze_platform_context, @@ -349,6 +351,104 @@ async def test_event_reply_builds_message_chain_for_the_frozen_target() -> None: assert call.kwargs['message'][0].text == 'hello' +@pytest.mark.asyncio +async def test_present_interactive_card_uses_host_interaction_manager() -> None: + adapter = SimpleNamespace(get_supported_apis=lambda: ['interaction.request']) + manager = SimpleNamespace( + present_tool_request=AsyncMock( + return_value={'ok': True, 'status': 'pending', 'interaction_id': 'card-host-generated'} + ) + ) + ap = SimpleNamespace( + agent_run_orchestrator=SimpleNamespace(interaction_manager=manager), + platform_mgr=SimpleNamespace(get_bot_by_uuid=AsyncMock(return_value=SimpleNamespace(adapter=adapter))), + ) + event = _event('message.received') + event.delivery.platform_capabilities['supported_apis'].append('interaction.request') + session = { + 'run_id': 'run-1', + 'runner_id': 'plugin:test/Agent/default', + 'authorization': { + 'bot_id': 'bot-1', + 'platform_context': freeze_platform_context(event), + }, + } + + result = await execute_platform_tool( + ap, + object(), + session, + PRESENT_INTERACTIVE_CARD_TOOL, + { + 'kind': 'choice', + 'title': 'Choose deployment environment', + 'fields': [ + { + 'id': 'environment', + 'label': 'Environment', + 'type': 'select', + 'options': [ + {'value': 'staging', 'label': 'Staging'}, + {'value': 'production', 'label': 'Production'}, + ], + } + ], + }, + ) + + assert result == {'ok': True, 'status': 'pending', 'interaction_id': 'card-host-generated'} + payload = manager.present_tool_request.await_args.kwargs['request_payload'] + assert payload['interaction_id'].startswith('card-') + assert payload['fallback_text'].startswith('Choose deployment environment') + assert 'callback_token' not in payload + assert manager.present_tool_request.await_args.kwargs['session'] is session + assert manager.present_tool_request.await_args.kwargs['adapter'] is adapter + + with pytest.raises(ValueError, match='cannot combine fields and actions'): + await execute_platform_tool( + ap, + object(), + session, + PRESENT_INTERACTIVE_CARD_TOOL, + { + 'title': 'Invalid mixed step', + 'fields': [ + { + 'id': 'environment', + 'label': 'Environment', + 'type': 'select', + 'options': [{'value': 'staging', 'label': 'Staging'}], + } + ], + 'actions': [{'id': 'continue', 'label': 'Continue'}], + }, + ) + + +def test_interactive_card_tool_is_automatic_only_for_interactive_message_surfaces() -> None: + selected = resolve_agent_platform_tool_names({}, 'message.received') + assert PRESENT_INTERACTIVE_CARD_TOOL in selected + assert PRESENT_INTERACTIVE_CARD_TOOL not in resolve_agent_platform_tool_names({}, 'group.member_joined') + + +def test_interactive_card_tool_schema_is_narrowed_to_adapter_capabilities() -> None: + event = _event('message.received') + event.delivery.platform_capabilities['supported_apis'].append('interaction.request') + event.delivery.interactions = InteractionDeliveryCapabilities( + field_types=['select'], + action_styles=['default', 'primary'], + supports_updates=True, + max_fields=1, + ) + + resources, _ = build_platform_tool_resources(event, [PRESENT_INTERACTIVE_CARD_TOOL], ['call']) + + parameters = resources[0]['parameters']['properties'] + assert parameters['fields']['maxItems'] == 1 + assert parameters['fields']['items']['properties']['type']['enum'] == ['select'] + assert parameters['actions']['items']['properties']['style']['enum'] == ['default', 'primary'] + + @pytest.mark.asyncio async def test_platform_action_rejects_parameters_outside_the_declared_schema() -> None: adapter = SimpleNamespace(