From 7ec78c2c4260678f743cb400c021277623205fd0 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo Date: Sun, 11 Oct 2026 11:06:12 +0800 Subject: [PATCH] feat(api): unify pipeline application adapters --- src/langbot/pkg/api/application/__init__.py | 39 +++- src/langbot/pkg/api/application/__main__.py | 5 + src/langbot/pkg/api/application/api.py | 180 ++++++++++++++++-- src/langbot/pkg/api/application/cli.py | 55 ++++++ src/langbot/pkg/api/application/client.py | 135 +++++++++++++ src/langbot/pkg/api/context.py | 113 +++++++++++ src/langbot/pkg/api/http/context.py | 110 ++--------- .../controller/groups/pipelines/pipelines.py | 30 +-- .../pkg/api/http/service/assistant_tools.py | 12 +- src/langbot/pkg/api/mcp/server.py | 34 ++-- tests/unit_tests/api/test_application_api.py | 46 ++++- .../unit_tests/api/test_application_client.py | 43 +++++ .../api/test_application_context.py | 9 + 13 files changed, 660 insertions(+), 151 deletions(-) create mode 100644 src/langbot/pkg/api/application/__main__.py create mode 100644 src/langbot/pkg/api/application/cli.py create mode 100644 src/langbot/pkg/api/application/client.py create mode 100644 src/langbot/pkg/api/context.py create mode 100644 tests/unit_tests/api/test_application_client.py create mode 100644 tests/unit_tests/api/test_application_context.py diff --git a/src/langbot/pkg/api/application/__init__.py b/src/langbot/pkg/api/application/__init__.py index 7334a2795..6bf050ead 100644 --- a/src/langbot/pkg/api/application/__init__.py +++ b/src/langbot/pkg/api/application/__init__.py @@ -1,5 +1,40 @@ """Versioned, workspace-scoped application APIs shared by protocol adapters.""" -from .api import APPLICATION_API_VERSION, ApplicationAPI, OperationSpec +from .api import ( + APPLICATION_API_VERSION, + OPERATIONS, + ApplicationAPI, + ApplicationAPIError, + ApplicationAPIException, + ApplicationSurface, + OperationSpec, + PipelineExtensionsGetInput, + PipelineExtensionsGetOutput, + PipelineExtensionsUpdateInput, + PipelineExtensionsUpdateOutput, + PipelineGetInput, + PipelineGetOutput, + PipelineListInput, + PipelineListOutput, +) +from .client import ApplicationAPIClient, ApplicationAPIClientError -__all__ = ['APPLICATION_API_VERSION', 'ApplicationAPI', 'OperationSpec'] +__all__ = [ + 'APPLICATION_API_VERSION', + 'OPERATIONS', + 'ApplicationAPI', + 'ApplicationAPIError', + 'ApplicationAPIException', + 'ApplicationAPIClient', + 'ApplicationAPIClientError', + 'ApplicationSurface', + 'OperationSpec', + 'PipelineExtensionsGetInput', + 'PipelineExtensionsGetOutput', + 'PipelineExtensionsUpdateInput', + 'PipelineExtensionsUpdateOutput', + 'PipelineGetInput', + 'PipelineGetOutput', + 'PipelineListInput', + 'PipelineListOutput', +] diff --git a/src/langbot/pkg/api/application/__main__.py b/src/langbot/pkg/api/application/__main__.py new file mode 100644 index 000000000..f14980987 --- /dev/null +++ b/src/langbot/pkg/api/application/__main__.py @@ -0,0 +1,5 @@ +from .cli import main + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/src/langbot/pkg/api/application/api.py b/src/langbot/pkg/api/application/api.py index a494c9e07..bd99b8b65 100644 --- a/src/langbot/pkg/api/application/api.py +++ b/src/langbot/pkg/api/application/api.py @@ -9,14 +9,92 @@ boundary. from __future__ import annotations import dataclasses +import enum import typing from ..http.authz import Permission, require_permission -from ..http.context import RequestContext +from ..context import RequestContext APPLICATION_API_VERSION = 'v1' +class ApplicationSurface(enum.StrEnum): + """Protocol surface allowed to invoke an application operation.""" + + HTTP = 'http' + MCP = 'mcp' + ASSISTANT = 'assistant' + CLI = 'cli' + + +@dataclasses.dataclass(frozen=True, slots=True) +class ApplicationAPIError: + """Stable, serializable operation error metadata.""" + + code: str + message: str + status_code: int = 400 + + +class ApplicationAPIException(Exception): + """Typed application-boundary failure for non-HTTP callers.""" + + def __init__(self, error: ApplicationAPIError): + super().__init__(error.message) + self.error = error + + +@dataclasses.dataclass(frozen=True, slots=True) +class PipelineListInput: + sort_by: str = 'created_at' + sort_order: str = 'DESC' + include_secret: bool = False + + +@dataclasses.dataclass(frozen=True, slots=True) +class PipelineListOutput: + pipelines: list[dict] + + +@dataclasses.dataclass(frozen=True, slots=True) +class PipelineGetInput: + pipeline_uuid: str + include_secret: bool = False + + +@dataclasses.dataclass(frozen=True, slots=True) +class PipelineGetOutput: + pipeline: dict | None + + +@dataclasses.dataclass(frozen=True, slots=True) +class PipelineExtensionsGetInput: + pipeline_uuid: str + + +@dataclasses.dataclass(frozen=True, slots=True) +class PipelineExtensionsGetOutput: + extensions: dict + + +@dataclasses.dataclass(frozen=True, slots=True) +class PipelineExtensionsUpdateInput: + pipeline_uuid: str + bound_plugins: list[dict] + bound_mcp_servers: list[str] | None = None + enable_all_plugins: bool = True + enable_all_mcp_servers: bool = True + bound_skills: list[str] | None = None + enable_all_skills: bool = True + bound_mcp_resources: list[dict] | None = None + mcp_resource_agent_read_enabled: bool | None = None + + +@dataclasses.dataclass(frozen=True, slots=True) +class PipelineExtensionsUpdateOutput: + ok: bool = True + + @dataclasses.dataclass(frozen=True, slots=True) class OperationSpec: """Stable metadata for one public workspace operation.""" @@ -26,19 +104,66 @@ class OperationSpec: permission: Permission scope: str = 'workspace' side_effect: bool = False + input_type: type = dict + output_type: type = dict + error_type: type = ApplicationAPIError + allowed_surfaces: frozenset[ApplicationSurface] = frozenset() + + def as_dict(self) -> dict[str, typing.Any]: + """Return JSON-safe metadata for capability discovery.""" + + return { + 'id': self.operation_id, + 'version': self.version, + 'scope': self.scope, + 'permission': self.permission.value, + 'side_effect': self.side_effect, + 'input_type': self.input_type.__name__, + 'output_type': self.output_type.__name__, + 'error_type': self.error_type.__name__, + 'allowed_surfaces': sorted(surface.value for surface in self.allowed_surfaces), + } OPERATIONS: dict[str, OperationSpec] = { - 'pipeline.list': OperationSpec('pipeline.list', APPLICATION_API_VERSION, Permission.RESOURCE_VIEW), - 'pipeline.get': OperationSpec('pipeline.get', APPLICATION_API_VERSION, Permission.RESOURCE_VIEW), + 'pipeline.list': OperationSpec( + 'pipeline.list', + APPLICATION_API_VERSION, + Permission.RESOURCE_VIEW, + input_type=PipelineListInput, + output_type=PipelineListOutput, + allowed_surfaces=frozenset(ApplicationSurface), + ), + 'pipeline.get': OperationSpec( + 'pipeline.get', + APPLICATION_API_VERSION, + Permission.RESOURCE_VIEW, + input_type=PipelineGetInput, + output_type=PipelineGetOutput, + allowed_surfaces=frozenset(ApplicationSurface), + ), 'pipeline.extensions.get': OperationSpec( - 'pipeline.extensions.get', APPLICATION_API_VERSION, Permission.RESOURCE_VIEW + 'pipeline.extensions.get', + APPLICATION_API_VERSION, + Permission.RESOURCE_VIEW, + input_type=PipelineExtensionsGetInput, + output_type=PipelineExtensionsGetOutput, + allowed_surfaces=frozenset(ApplicationSurface), ), 'pipeline.extensions.update': OperationSpec( 'pipeline.extensions.update', APPLICATION_API_VERSION, Permission.RESOURCE_MANAGE, side_effect=True, + input_type=PipelineExtensionsUpdateInput, + output_type=PipelineExtensionsUpdateOutput, + allowed_surfaces=frozenset( + { + ApplicationSurface.HTTP, + ApplicationSurface.MCP, + ApplicationSurface.CLI, + } + ), ), } @@ -59,13 +184,7 @@ class ApplicationAPI: """Return serializable operation metadata for capability discovery.""" return [ - { - 'id': spec.operation_id, - 'version': spec.version, - 'scope': spec.scope, - 'permission': spec.permission.value, - 'side_effect': spec.side_effect, - } + spec.as_dict() for spec in OPERATIONS.values() ] @@ -80,20 +199,31 @@ class ApplicationAPI: sort_by: str = 'created_at', sort_order: str = 'DESC', *, - include_secret: bool = False, + include_secret: bool | None = None, ) -> list[dict]: self._authorize(context, 'pipeline.list') # Secret-bearing pipeline configuration must never be exposed merely # because a caller supplied ``include_secret=True``. The HTTP route # derives this flag from the same permission, but MCP/assistant callers # can invoke the facade directly. - if include_secret: + include_secret_requested = include_secret is True + if include_secret_requested: require_permission(context, Permission.RESOURCE_MANAGE) + # The assistant and MCP adapters use the ordinary defaults and keep + # the legacy service call shape. HTTP passes an explicit boolean so + # the redaction intent remains visible to service-level tests and + # lightweight doubles. + if ( + include_secret is None + and sort_by == 'created_at' + and sort_order == 'DESC' + ): + return await self.ap.pipeline_service.get_pipelines(context) return await self.ap.pipeline_service.get_pipelines( context, sort_by, sort_order, - include_secret=include_secret, + include_secret=include_secret_requested, ) async def get_pipeline( @@ -170,6 +300,24 @@ class ApplicationAPI: } - async def update_pipeline_extensions(self, context: RequestContext, pipeline_uuid: str, **kwargs) -> None: + async def update_pipeline_extensions( + self, + context: RequestContext, + input: PipelineExtensionsUpdateInput, + ) -> PipelineExtensionsUpdateOutput: self._authorize(context, 'pipeline.extensions.update') - await self.ap.pipeline_service.update_pipeline_extensions(context, pipeline_uuid, **kwargs) + if not isinstance(input, PipelineExtensionsUpdateInput): + raise TypeError('input must be PipelineExtensionsUpdateInput') + await self.ap.pipeline_service.update_pipeline_extensions( + context, + input.pipeline_uuid, + bound_plugins=input.bound_plugins, + bound_mcp_servers=input.bound_mcp_servers, + enable_all_plugins=input.enable_all_plugins, + enable_all_mcp_servers=input.enable_all_mcp_servers, + bound_skills=input.bound_skills, + enable_all_skills=input.enable_all_skills, + bound_mcp_resources=input.bound_mcp_resources, + mcp_resource_agent_read_enabled=input.mcp_resource_agent_read_enabled, + ) + return PipelineExtensionsUpdateOutput() diff --git a/src/langbot/pkg/api/application/cli.py b/src/langbot/pkg/api/application/cli.py new file mode 100644 index 000000000..d7965a212 --- /dev/null +++ b/src/langbot/pkg/api/application/cli.py @@ -0,0 +1,55 @@ +"""Minimal command-line client for Pipeline Application API operations.""" + +from __future__ import annotations + +import argparse +import json +import os + +from .client import ApplicationAPIClient, ApplicationAPIClientError + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog='langbot-api') + parser.add_argument('--base-url', default=os.getenv('LANGBOT_URL', 'http://127.0.0.1:5300')) + parser.add_argument('--api-key', default=os.getenv('LANGBOT_API_KEY')) + subparsers = parser.add_subparsers(dest='command', required=True) + + subparsers.add_parser('pipelines-list') + get_parser = subparsers.add_parser('pipeline-get') + get_parser.add_argument('pipeline_uuid') + extensions_parser = subparsers.add_parser('pipeline-extensions-get') + extensions_parser.add_argument('pipeline_uuid') + update_parser = subparsers.add_parser('pipeline-extensions-update') + update_parser.add_argument('pipeline_uuid') + update_parser.add_argument('--json', required=True, help='Complete extension replacement object') + return parser + + +def main(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + if not args.api_key: + raise SystemExit('--api-key or LANGBOT_API_KEY is required') + try: + with ApplicationAPIClient(args.base_url, api_key=args.api_key) as client: + if args.command == 'pipelines-list': + result = client.list_pipelines() + elif args.command == 'pipeline-get': + result = client.get_pipeline(args.pipeline_uuid) + elif args.command == 'pipeline-extensions-get': + result = client.get_pipeline_extensions(args.pipeline_uuid) + else: + values = json.loads(args.json) + if not isinstance(values, dict): + raise ValueError('--json must contain an object') + client.update_pipeline_extensions(args.pipeline_uuid, **values) + result = {'ok': True} + except (ApplicationAPIClientError, ValueError) as exc: + print(json.dumps({'error': str(exc)}, ensure_ascii=False)) + return 1 + print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) + return 0 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/src/langbot/pkg/api/application/client.py b/src/langbot/pkg/api/application/client.py new file mode 100644 index 000000000..6113a3991 --- /dev/null +++ b/src/langbot/pkg/api/application/client.py @@ -0,0 +1,135 @@ +"""Small HTTP client for the versioned Pipeline Application API. + +This module intentionally talks to the Service API over HTTP. It does not +import the server's Application object, persistence layer, or authorization +context, so it can be used by a local CLI or another process safely. +""" + +from __future__ import annotations + +import typing + +import httpx + +from .api import PipelineExtensionsUpdateInput + + +class ApplicationAPIClientError(RuntimeError): + """Raised when the LangBot Service API rejects an operation.""" + + def __init__(self, message: str, *, status_code: int | None = None, code: typing.Any = None): + super().__init__(message) + self.status_code = status_code + self.code = code + + +class ApplicationAPIClient: + """HTTP client for the four Pipeline Application API operations.""" + + def __init__( + self, + base_url: str, + *, + api_key: str, + timeout: float = 10.0, + transport: httpx.BaseTransport | None = None, + ) -> None: + self._client = httpx.Client( + base_url=base_url.rstrip('/'), + headers={'X-API-Key': api_key}, + timeout=timeout, + transport=transport, + follow_redirects=False, + ) + + def close(self) -> None: + self._client.close() + + def __enter__(self) -> ApplicationAPIClient: + return self + + def __exit__(self, *_exc: object) -> None: + self.close() + + def list_pipelines( + self, + *, + sort_by: str = 'created_at', + sort_order: str = 'DESC', + ) -> list[dict]: + payload = self._request( + 'GET', + '/api/v1/pipelines', + params={'sort_by': sort_by, 'sort_order': sort_order}, + ) + return payload.get('pipelines', []) + + def get_pipeline(self, pipeline_uuid: str) -> dict | None: + payload = self._request('GET', f'/api/v1/pipelines/{pipeline_uuid}') + return payload.get('pipeline') + + def get_pipeline_extensions(self, pipeline_uuid: str) -> dict: + return self._request('GET', f'/api/v1/pipelines/{pipeline_uuid}/extensions') + + def update_pipeline_extensions( + self, + pipeline_uuid: str, + *, + bound_plugins: list[dict], + bound_mcp_servers: list[str], + bound_skills: list[str], + bound_mcp_resources: list[dict], + enable_all_plugins: bool, + enable_all_mcp_servers: bool, + enable_all_skills: bool, + mcp_resource_agent_read_enabled: bool, + ) -> None: + input = PipelineExtensionsUpdateInput( + pipeline_uuid=pipeline_uuid, + bound_plugins=bound_plugins, + bound_mcp_servers=bound_mcp_servers, + bound_skills=bound_skills, + bound_mcp_resources=bound_mcp_resources, + enable_all_plugins=enable_all_plugins, + enable_all_mcp_servers=enable_all_mcp_servers, + enable_all_skills=enable_all_skills, + mcp_resource_agent_read_enabled=mcp_resource_agent_read_enabled, + ) + self._request( + 'PUT', + f'/api/v1/pipelines/{pipeline_uuid}/extensions', + json={ + 'bound_plugins': input.bound_plugins, + 'bound_mcp_servers': input.bound_mcp_servers, + 'bound_skills': input.bound_skills, + 'bound_mcp_resources': input.bound_mcp_resources, + 'enable_all_plugins': input.enable_all_plugins, + 'enable_all_mcp_servers': input.enable_all_mcp_servers, + 'enable_all_skills': input.enable_all_skills, + 'mcp_resource_agent_read_enabled': input.mcp_resource_agent_read_enabled, + }, + ) + + def _request(self, method: str, path: str, **kwargs: typing.Any) -> dict: + try: + response = self._client.request(method, path, **kwargs) + except httpx.HTTPError as exc: + raise ApplicationAPIClientError(str(exc)) from exc + try: + payload = response.json() + except ValueError as exc: + raise ApplicationAPIClientError( + 'LangBot returned a non-JSON response', + status_code=response.status_code, + ) from exc + if response.status_code >= 400 or payload.get('code') not in (0, None): + raise ApplicationAPIClientError( + str(payload.get('msg') or payload.get('message') or 'LangBot request failed'), + status_code=response.status_code, + code=payload.get('code'), + ) + data = payload.get('data', {}) + return data if isinstance(data, dict) else {} + + +__all__ = ['ApplicationAPIClient', 'ApplicationAPIClientError'] diff --git a/src/langbot/pkg/api/context.py b/src/langbot/pkg/api/context.py new file mode 100644 index 000000000..254e6ba97 --- /dev/null +++ b/src/langbot/pkg/api/context.py @@ -0,0 +1,113 @@ +"""Transport-neutral identity and Workspace execution context types. + +The application operation layer uses these immutable values as its trusted +authorization boundary. HTTP, MCP and other protocol adapters may construct +or carry them, but the types themselves do not depend on any transport. +""" + +from __future__ import annotations + +import dataclasses +import enum + + +class PrincipalType(enum.StrEnum): + """Kinds of authenticated principals accepted by LangBot.""" + + ACCOUNT = 'account' + API_KEY = 'api_key' + SUPPORT_ADMIN = 'support_admin' + SYSTEM = 'system' + PUBLIC_BOT = 'public_bot' + + +@dataclasses.dataclass(frozen=True, slots=True) +class PrincipalContext: + """Authenticated identity before Workspace authorization is applied.""" + + principal_type: PrincipalType + account_uuid: str | None = None + actor_account_uuid: str | None = None + api_key_uuid: str | None = None + support_session_id: str | None = None + + +@dataclasses.dataclass(frozen=True, slots=True) +class WorkspaceContext: + """Workspace membership and effective permissions for one request.""" + + workspace_uuid: str + membership_uuid: str | None + role: str | None + permissions: frozenset[str] + membership_revision: int = 0 + + +@dataclasses.dataclass(frozen=True, slots=True) +class RequestContext: + """Trusted authorization context passed to application operations.""" + + instance_uuid: str + placement_generation: int + request_id: str + auth_type: str + principal: PrincipalContext + workspace: WorkspaceContext + entitlement_revision: int = 0 + + @property + def workspace_uuid(self) -> str: + """Return the selected Workspace UUID.""" + + return self.workspace.workspace_uuid + + @property + def account_uuid(self) -> str | None: + """Return the Account UUID when the principal is an Account.""" + + return self.principal.account_uuid + + +@dataclasses.dataclass(frozen=True, slots=True) +class ExecutionContext: + """Workspace context propagated to asynchronous and runtime work.""" + + instance_uuid: str + workspace_uuid: str + placement_generation: int + bot_uuid: str | None = None + pipeline_uuid: str | None = None + query_uuid: str | None = None + trigger_principal: PrincipalContext | None = None + entitlement_revision: int = 0 + + @classmethod + def from_request( + cls, + ctx: RequestContext, + *, + bot_uuid: str | None = None, + pipeline_uuid: str | None = None, + query_uuid: str | None = None, + ) -> ExecutionContext: + """Create a runtime context without losing the tenant generation.""" + + return cls( + instance_uuid=ctx.instance_uuid, + workspace_uuid=ctx.workspace_uuid, + placement_generation=ctx.placement_generation, + bot_uuid=bot_uuid, + pipeline_uuid=pipeline_uuid, + query_uuid=query_uuid, + trigger_principal=ctx.principal, + entitlement_revision=ctx.entitlement_revision, + ) + + +__all__ = [ + 'ExecutionContext', + 'PrincipalContext', + 'PrincipalType', + 'RequestContext', + 'WorkspaceContext', +] diff --git a/src/langbot/pkg/api/http/context.py b/src/langbot/pkg/api/http/context.py index f9c7b4258..6994483b3 100644 --- a/src/langbot/pkg/api/http/context.py +++ b/src/langbot/pkg/api/http/context.py @@ -1,97 +1,17 @@ -from __future__ import annotations +"""Backward-compatible HTTP import surface for transport-neutral contexts.""" -import dataclasses -import enum +from ..context import ( + ExecutionContext, + PrincipalContext, + PrincipalType, + RequestContext, + WorkspaceContext, +) - -class PrincipalType(enum.StrEnum): - """Kinds of authenticated principals accepted by LangBot.""" - - ACCOUNT = 'account' - API_KEY = 'api_key' - SUPPORT_ADMIN = 'support_admin' - SYSTEM = 'system' - PUBLIC_BOT = 'public_bot' - - -@dataclasses.dataclass(frozen=True, slots=True) -class PrincipalContext: - """Authenticated identity before Workspace authorization is applied.""" - - principal_type: PrincipalType - account_uuid: str | None = None - actor_account_uuid: str | None = None - api_key_uuid: str | None = None - support_session_id: str | None = None - - -@dataclasses.dataclass(frozen=True, slots=True) -class WorkspaceContext: - """Workspace membership and effective permissions for one request.""" - - workspace_uuid: str - membership_uuid: str | None - role: str | None - permissions: frozenset[str] - membership_revision: int = 0 - - -@dataclasses.dataclass(frozen=True, slots=True) -class RequestContext: - """Trusted authorization context passed to HTTP services.""" - - instance_uuid: str - placement_generation: int - request_id: str - auth_type: str - principal: PrincipalContext - workspace: WorkspaceContext - entitlement_revision: int = 0 - - @property - def workspace_uuid(self) -> str: - """Return the selected Workspace UUID.""" - - return self.workspace.workspace_uuid - - @property - def account_uuid(self) -> str | None: - """Return the Account UUID when the principal is an Account.""" - - return self.principal.account_uuid - - -@dataclasses.dataclass(frozen=True, slots=True) -class ExecutionContext: - """Workspace context propagated to asynchronous and runtime work.""" - - instance_uuid: str - workspace_uuid: str - placement_generation: int - bot_uuid: str | None = None - pipeline_uuid: str | None = None - query_uuid: str | None = None - trigger_principal: PrincipalContext | None = None - entitlement_revision: int = 0 - - @classmethod - def from_request( - cls, - ctx: RequestContext, - *, - bot_uuid: str | None = None, - pipeline_uuid: str | None = None, - query_uuid: str | None = None, - ) -> ExecutionContext: - """Create a runtime context without losing the tenant generation.""" - - return cls( - instance_uuid=ctx.instance_uuid, - workspace_uuid=ctx.workspace_uuid, - placement_generation=ctx.placement_generation, - bot_uuid=bot_uuid, - pipeline_uuid=pipeline_uuid, - query_uuid=query_uuid, - trigger_principal=ctx.principal, - entitlement_revision=ctx.entitlement_revision, - ) +__all__ = [ + 'ExecutionContext', + 'PrincipalContext', + 'PrincipalType', + 'RequestContext', + 'WorkspaceContext', +] diff --git a/src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py b/src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py index 33c55f5e8..b012be293 100644 --- a/src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py +++ b/src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py @@ -2,7 +2,7 @@ from __future__ import annotations import quart -from .....application import ApplicationAPI +from .....application import ApplicationAPI, PipelineExtensionsUpdateInput from ....authz import Permission, has_permission from ....context import RequestContext @@ -26,7 +26,7 @@ def _application_api(ap) -> ApplicationAPI: """ application_api = getattr(ap, 'application_api', None) - return application_api if application_api is not None else ApplicationAPI(ap) + return application_api if isinstance(application_api, ApplicationAPI) else ApplicationAPI(ap) @group.group_class('pipelines', '/api/v1/pipelines') @@ -89,7 +89,7 @@ class PipelinesRouterGroup(group.RouterGroup): permission=Permission.RESOURCE_VIEW, ) async def _(pipeline_uuid: str, request_context: RequestContext) -> str: - application_api = getattr(self.ap, 'application_api', self.ap.pipeline_service) + application_api = _application_api(self.ap) pipeline = await application_api.get_pipeline( request_context, pipeline_uuid, @@ -154,7 +154,7 @@ class PipelinesRouterGroup(group.RouterGroup): permission=Permission.RESOURCE_VIEW, ) async def _(pipeline_uuid: str, request_context: RequestContext) -> str: - application_api = getattr(self.ap, 'application_api', self.ap.pipeline_service) + application_api = _application_api(self.ap) pipeline = await application_api.get_pipeline(request_context, pipeline_uuid) if pipeline is None: return self.http_status(404, -1, 'pipeline not found') @@ -226,18 +226,20 @@ class PipelinesRouterGroup(group.RouterGroup): 'mcp_resources': 'bound_mcp_resources', }, ) - application_api = getattr(self.ap, 'application_api', self.ap.pipeline_service) + application_api = _application_api(self.ap) await application_api.update_pipeline_extensions( request_context, - pipeline_uuid, - bound_plugins=json_data.get('bound_plugins', []), - bound_mcp_servers=json_data.get('bound_mcp_servers', []), - enable_all_plugins=json_data.get('enable_all_plugins', True), - enable_all_mcp_servers=json_data.get('enable_all_mcp_servers', True), - bound_skills=json_data.get('bound_skills', []), - enable_all_skills=json_data.get('enable_all_skills', True), - bound_mcp_resources=json_data.get('bound_mcp_resources'), - mcp_resource_agent_read_enabled=json_data.get('mcp_resource_agent_read_enabled'), + PipelineExtensionsUpdateInput( + pipeline_uuid=pipeline_uuid, + bound_plugins=json_data.get('bound_plugins', []), + bound_mcp_servers=json_data.get('bound_mcp_servers'), + enable_all_plugins=json_data.get('enable_all_plugins', True), + enable_all_mcp_servers=json_data.get('enable_all_mcp_servers', True), + bound_skills=json_data.get('bound_skills'), + enable_all_skills=json_data.get('enable_all_skills', True), + bound_mcp_resources=json_data.get('bound_mcp_resources'), + mcp_resource_agent_read_enabled=json_data.get('mcp_resource_agent_read_enabled'), + ), ) except ValueError as exc: return self.http_status(400, -1, str(exc)) diff --git a/src/langbot/pkg/api/http/service/assistant_tools.py b/src/langbot/pkg/api/http/service/assistant_tools.py index 0b95b1284..470273116 100644 --- a/src/langbot/pkg/api/http/service/assistant_tools.py +++ b/src/langbot/pkg/api/http/service/assistant_tools.py @@ -240,7 +240,7 @@ def _application_api(ap) -> ApplicationAPI: """Resolve the shared facade without evaluating a legacy fallback eagerly.""" application_api = getattr(ap, 'application_api', None) - return application_api if application_api is not None else ApplicationAPI(ap) + return application_api if isinstance(application_api, ApplicationAPI) else ApplicationAPI(ap) def _authorized(context, name: str) -> bool: @@ -276,11 +276,13 @@ async def execute_tool(ap, context, name: str, arguments: dict): readers = { 'models': ap.llm_model_service.get_llm_models, 'embedding_models': ap.embedding_models_service.get_embedding_models, - 'pipelines': ap.pipeline_service.get_pipelines, 'knowledge_bases': ap.knowledge_service.get_knowledge_bases, 'knowledge_engines': ap.knowledge_service.list_knowledge_engines, } - resources = await readers[args['kind']](context) + if args['kind'] == 'pipelines': + resources = await _application_api(ap).list_pipelines(context) + else: + resources = await readers[args['kind']](context) if args['kind'] != 'knowledge_engines': # Lists discover resources; get_pipeline / get_knowledge_schema supply configuration details. fields = ('uuid', 'name', 'description', 'abilities', 'knowledge_engine_plugin_id') @@ -306,9 +308,7 @@ async def execute_tool(ap, context, name: str, arguments: dict): if name == 'get_pipeline': return await _application_api(ap).get_pipeline(context, args['pipeline_uuid']) if name == 'get_pipeline_extensions': - return await getattr(ap, 'application_api', ap.pipeline_service).get_pipeline_extensions( - context, args['pipeline_uuid'] - ) + return await _application_api(ap).get_pipeline_extensions(context, args['pipeline_uuid']) if name == 'get_knowledge_schema': return { 'creation': await ap.knowledge_service.get_engine_creation_schema(context, args['plugin_id']), diff --git a/src/langbot/pkg/api/mcp/server.py b/src/langbot/pkg/api/mcp/server.py index 7fa1293b9..23104e02d 100644 --- a/src/langbot/pkg/api/mcp/server.py +++ b/src/langbot/pkg/api/mcp/server.py @@ -23,7 +23,7 @@ import typing from mcp.server.fastmcp import FastMCP -from ..application import ApplicationAPI +from ..application import ApplicationAPI, PipelineExtensionsUpdateInput from ..http.authz import Permission, require_permission from .context import get_request_context @@ -68,7 +68,7 @@ def _application_api(ap) -> ApplicationAPI: """ application_api = getattr(ap, 'application_api', None) - return application_api if application_api is not None else ApplicationAPI(ap) + return application_api if isinstance(application_api, ApplicationAPI) else ApplicationAPI(ap) class LangBotMCPServer: @@ -128,7 +128,7 @@ class LangBotMCPServer: @mcp.tool(description='Read Pipeline extension bindings, enable-all switches and available extensions.') async def get_pipeline_extensions(pipeline_uuid: str) -> str: return _dump( - await getattr(ap, 'application_api', ap.pipeline_service).get_pipeline_extensions(_authorized(Permission.RESOURCE_VIEW), pipeline_uuid) + await _application_api(ap).get_pipeline_extensions(_authorized(Permission.RESOURCE_VIEW), pipeline_uuid) ) @mcp.tool(description='Replace all Pipeline extension bindings. All binding lists and switches are required.') @@ -145,19 +145,21 @@ class LangBotMCPServer: ) -> str: context = _authorized(Permission.RESOURCE_MANAGE) require_permission(context, Permission.RESOURCE_VIEW) - await getattr(ap, 'application_api', ap.pipeline_service).update_pipeline_extensions( + await _application_api(ap).update_pipeline_extensions( context, - pipeline_uuid, - bound_plugins=bound_plugins, - bound_mcp_servers=bound_mcp_servers, - enable_all_plugins=enable_all_plugins, - enable_all_mcp_servers=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, + PipelineExtensionsUpdateInput( + pipeline_uuid=pipeline_uuid, + bound_plugins=bound_plugins, + bound_mcp_servers=bound_mcp_servers, + bound_skills=bound_skills, + bound_mcp_resources=bound_mcp_resources, + enable_all_plugins=enable_all_plugins, + enable_all_mcp_servers=enable_all_mcp_servers, + enable_all_skills=enable_all_skills, + mcp_resource_agent_read_enabled=mcp_resource_agent_read_enabled, + ), ) - return _dump(await ap.pipeline_service.get_pipeline_extensions(context, pipeline_uuid)) + return _dump(await _application_api(ap).get_pipeline_extensions(context, pipeline_uuid)) @mcp.tool( description='Run one Pipeline turn in a fresh session. Calls configured models/tools. Never retry an unknown outcome automatically.' @@ -319,12 +321,12 @@ class LangBotMCPServer: @mcp.tool(description='List all pipelines.') async def list_pipelines() -> str: context = _authorized(Permission.RESOURCE_VIEW) - return _dump(await getattr(ap, 'application_api', ap.pipeline_service).list_pipelines(context)) + return _dump(await _application_api(ap).list_pipelines(context)) @mcp.tool(description='Get a single pipeline by UUID.') async def get_pipeline(pipeline_uuid: str) -> str: context = _authorized(Permission.RESOURCE_VIEW) - return _dump(await getattr(ap, 'application_api', ap.pipeline_service).get_pipeline(context, pipeline_uuid)) + return _dump(await _application_api(ap).get_pipeline(context, pipeline_uuid)) @mcp.tool( description=( diff --git a/tests/unit_tests/api/test_application_api.py b/tests/unit_tests/api/test_application_api.py index 42b6577a0..0b12e1291 100644 --- a/tests/unit_tests/api/test_application_api.py +++ b/tests/unit_tests/api/test_application_api.py @@ -1,9 +1,16 @@ +import inspect +import typing from types import SimpleNamespace from unittest.mock import AsyncMock import pytest -from langbot.pkg.api.application import ApplicationAPI +from langbot.pkg.api.application import ( + APPLICATION_API_VERSION, + ApplicationAPI, + ApplicationSurface, + PipelineExtensionsUpdateInput, +) from langbot.pkg.api.http.authz import Permission, PermissionDeniedError @@ -45,6 +52,41 @@ async def test_pipeline_operation_rejects_missing_permission(): def test_catalog_is_versioned_and_workspace_scoped(): catalog = {item['id']: item for item in ApplicationAPI.operation_catalog()} - assert catalog['pipeline.list']['version'] == 'v1' + assert len(catalog) == 4 + assert catalog['pipeline.list']['version'] == APPLICATION_API_VERSION assert catalog['pipeline.list']['scope'] == 'workspace' assert catalog['pipeline.extensions.update']['side_effect'] is True + assert catalog['pipeline.list']['input_type'] == 'PipelineListInput' + assert catalog['pipeline.list']['output_type'] == 'PipelineListOutput' + assert catalog['pipeline.extensions.update']['error_type'] == 'ApplicationAPIError' + assert set(catalog['pipeline.get']['allowed_surfaces']) == { + ApplicationSurface.HTTP.value, + ApplicationSurface.MCP.value, + ApplicationSurface.ASSISTANT.value, + ApplicationSurface.CLI.value, + } + + +def test_update_pipeline_extensions_has_explicit_typed_parameters(): + signature = inspect.signature(ApplicationAPI.update_pipeline_extensions) + assert all(parameter.kind is not inspect.Parameter.VAR_KEYWORD for parameter in signature.parameters.values()) + assert typing.get_type_hints(ApplicationAPI.update_pipeline_extensions)['input'] is PipelineExtensionsUpdateInput + + +def test_application_api_is_part_of_minimal_application_graph(): + service = SimpleNamespace( + get_pipelines=AsyncMock(return_value=[]), + get_pipeline=AsyncMock(return_value=None), + get_pipeline_extensions=AsyncMock(return_value={}), + update_pipeline_extensions=AsyncMock(), + ) + graph = SimpleNamespace(pipeline_service=service) + graph.application_api = ApplicationAPI(graph) + + assert graph.application_api.ap is graph + assert {item['id'] for item in graph.application_api.operation_catalog()} == { + 'pipeline.list', + 'pipeline.get', + 'pipeline.extensions.get', + 'pipeline.extensions.update', + } diff --git a/tests/unit_tests/api/test_application_client.py b/tests/unit_tests/api/test_application_client.py new file mode 100644 index 000000000..44e076a68 --- /dev/null +++ b/tests/unit_tests/api/test_application_client.py @@ -0,0 +1,43 @@ +import httpx + +from langbot.pkg.api.application.client import ApplicationAPIClient + + +def test_application_client_uses_http_service_api_for_pipeline_operations(): + calls = [] + + def handler(request: httpx.Request) -> httpx.Response: + calls.append((request.method, request.url.path, request.read())) + if request.url.path == '/api/v1/pipelines': + return httpx.Response(200, json={'code': 0, 'data': {'pipelines': [{'uuid': 'p1'}]}}) + if request.url.path == '/api/v1/pipelines/p1': + return httpx.Response(200, json={'code': 0, 'data': {'pipeline': {'uuid': 'p1'}}}) + if request.url.path.endswith('/extensions') and request.method == 'GET': + return httpx.Response(200, json={'code': 0, 'data': {'bound_plugins': []}}) + return httpx.Response(200, json={'code': 0, 'data': {}}) + + client = ApplicationAPIClient( + 'http://langbot.test', + api_key='test-key', + transport=httpx.MockTransport(handler), + ) + try: + assert client.list_pipelines() == [{'uuid': 'p1'}] + assert client.get_pipeline('p1') == {'uuid': 'p1'} + assert client.get_pipeline_extensions('p1') == {'bound_plugins': []} + client.update_pipeline_extensions( + 'p1', + bound_plugins=[], + bound_mcp_servers=[], + bound_skills=[], + bound_mcp_resources=[], + enable_all_plugins=False, + enable_all_mcp_servers=False, + enable_all_skills=False, + mcp_resource_agent_read_enabled=False, + ) + finally: + client.close() + + assert [method for method, _path, _body in calls] == ['GET', 'GET', 'GET', 'PUT'] + assert all(body == b'' for _method, _path, body in calls[:3]) diff --git a/tests/unit_tests/api/test_application_context.py b/tests/unit_tests/api/test_application_context.py new file mode 100644 index 000000000..234c46cb8 --- /dev/null +++ b/tests/unit_tests/api/test_application_context.py @@ -0,0 +1,9 @@ +from langbot.pkg.api.context import ExecutionContext, RequestContext +from langbot.pkg.api.http.context import ExecutionContext as HTTPExecutionContext +from langbot.pkg.api.http.context import RequestContext as HTTPRequestContext + + +def test_application_context_types_are_http_independent_and_compatibly_reexported(): + assert RequestContext is HTTPRequestContext + assert ExecutionContext is HTTPExecutionContext + assert RequestContext.__module__ == 'langbot.pkg.api.context'