Compare commits

..

6 Commits

Author SHA1 Message Date
huanghuoguoguo 9d877b41c2 test(skills): clarify manual QA perf gates 2026-06-25 20:46:31 +08:00
huanghuoguoguo 9b0f5b36f3 test(skills): add debug chat timing and isolation probes 2026-06-25 13:34:30 +08:00
huanghuoguoguo 7e36869494 test(skills): extend fake provider load profiles 2026-06-25 12:54:08 +08:00
huanghuoguoguo d59b49ec55 test(skills): add debug chat load gate 2026-06-25 11:48:59 +08:00
huanghuoguoguo 8749a9b56f test(skills): prepare user path performance gate 2026-06-25 10:07:04 +08:00
huanghuoguoguo 67437c2f5a Add performance and reliability QA gates 2026-06-25 00:07:37 +08:00
175 changed files with 1577 additions and 14613 deletions
+124 -69
View File
@@ -1,105 +1,160 @@
# AGENTS.md
This file guides code agents working in the LangBot main repository. `CLAUDE.md` is a symlink to this file.
This file guides code agents (Claude Code, GitHub Copilot, OpenAI Codex, etc.) working in the LangBot project. `CLAUDE.md` is a symlink to this file.
Read `ARCHITECTURE.md` before non-trivial backend, frontend, runtime, plugin, Box, MCP, persistence, or cross-repo SDK changes. This file is the working checklist; `ARCHITECTURE.md` is the system map.
## Project Overview
## Quick Facts
LangBot is an open-source, LLM-native instant-messaging bot development platform. It aims to provide an out-of-the-box IM bot development experience with Agent, RAG, MCP and other LLM application capabilities, supporting mainstream global IM platforms and exposing rich APIs for custom development.
- Python backend: `>=3.11,<4.0`, dependencies managed by `uv`.
- Frontend: `web/` is Vite + React Router 7 + shadcn/ui + Tailwind, managed by `pnpm`.
- Backend framework: Quart served by Hypercorn on `api.port`, default `5300`.
- Frontend dev server: `web/` on `3000`, with `VITE_API_BASE_URL` pointing at the backend.
- Plugin/Box/runtime contracts live in sibling repo `langbot-plugin-sdk`, pinned as `langbot-plugin` in `pyproject.toml`.
LangBot has a comprehensive web frontend — almost every operation can be performed through it.
## Essential Commands
- **Python**: `>=3.11,<4.0`, dependencies managed by `uv`. Package version is in `pyproject.toml`.
- **Frontend**: `web/` is a **Vite + React Router 7 + shadcn/ui + Tailwind CSS** SPA, managed by `pnpm`. (Note: this is NOT Next.js — the `dev` script is `vite`.)
- **Backend framework**: Quart (the async flavour of Flask). The HTTP API and the pre-built web UI are both served by the backend on `http://127.0.0.1:5300`.
## Repository Layout
```
LangBot/
├── main.py # Entrypoint shim -> langbot.__main__.main()
├── pyproject.toml # Python project + deps (uv), pins langbot-plugin==<x.y.z>
├── src/langbot/
│ ├── __main__.py # Real entrypoint, CLI args (--standalone-runtime, --standalone-box, --debug)
│ ├── pkg/ # Core backend package
│ │ ├── api/ # HTTP API controllers + services (Quart)
│ │ ├── core/ # App bootstrap, stages, task manager
│ │ ├── platform/ # IM platform adapters, bot managers, session managers
│ │ ├── provider/ # LLM providers, requesters, tool providers
│ │ ├── pipeline/ # Pipelines, stages, query pool
│ │ ├── plugin/ # Bridge connecting LangBot to the plugin runtime (see below)
│ │ ├── box/ # Code-sandbox subsystem (Docker / nsjail / E2B backends)
│ │ ├── skill/ # Skill subsystem
│ │ ├── rag/ , vector/ # RAG + vector store
│ │ ├── command/ # Built-in commands
│ │ ├── persistence/ # ORM models + Alembic migrations (SQLite & PostgreSQL)
│ │ ├── storage/ # Object/file storage abstractions
│ │ ├── config/, entity/, discover/, utils/, telemetry/, survey/
│ ├── libs/ # Vendored SDKs (qq_official_api, wecom_api, etc.)
│ └── templates/ # Config/component templates (e.g. templates/config.yaml)
├── web/ # Frontend SPA (Vite + React Router 7 + shadcn + Tailwind)
└── docker/ # docker-compose deployment files
```
## Development Environment Setup
Full guide lives in the wiki: **["开发配置" / Dev Config](https://docs.langbot.app/zh/develop/dev-config)**. Summary:
### Backend
```bash
uv sync --dev
uv run main.py
uv run pre-commit install
pip install uv
uv sync --dev # uv creates a .venv/ for you; point your editor's interpreter at it
uv run main.py # serves API + web UI on http://127.0.0.1:5300
```
On first run the config file is generated at `data/config.yaml`. DB is SQLite by default (zero setup); PostgreSQL is supported. Migrations run automatically on startup.
### Frontend
Requires Node.js + [pnpm](https://pnpm.io/installation).
```bash
cd web
cp .env.example .env # Windows: copy .env.example .env
pnpm install
pnpm dev
pnpm build
pnpm dev # http://127.0.0.1:3000 (npm install / npm run dev also work)
```
Useful focused tests:
`pnpm dev` reads `VITE_API_BASE_URL` from `web/.env` so the dev frontend can reach the backend on port `5300`. In production the frontend is pre-built into static files served by the backend on the same origin.
### Code formatting
The repo runs lint + format checks in CI. Install the pre-commit hooks so the same checks run locally before each commit:
```bash
uv run pytest tests/unit_tests -q
uv run pytest tests/integration -q
uv run pytest tests/integration/persistence -q
uv run pytest tests/manual/mcp_smoke.py
cd web
pnpm lint
pnpm test:e2e
uv run pre-commit install
```
Run the narrowest useful test first, then broader checks when confidence is needed.
## Plugin System
## Where to Look
LangBot's plugin system (Plugin SDK, CLI `lbp`, Plugin Runtime, and the shared entity/API definitions) lives in a **separate repository**: [`langbot-plugin-sdk`](https://github.com/langbot-app/langbot-plugin-sdk). LangBot depends on it via the pinned `langbot-plugin` package in `pyproject.toml`.
- Architecture map: `ARCHITECTURE.md`.
- Dev environment guide: https://docs.langbot.app/zh/develop/dev-config.
- Plugin runtime / CLI / SDK debugging: https://docs.langbot.app/zh/develop/plugin-runtime.
- API-key auth: `docs/API_KEY_AUTH.md`.
- Box deep-dive notes: `docs/review/box-architecture.md` and related files.
- In-repo skills: `skills/` is the single source of truth for LangBot agent skills.
- SDK repo: `../langbot-plugin-sdk/` when changing shared entities, plugin APIs, action protocol, `lbp rt`, or `lbp box`.
### Architecture (what to know inside this repo)
## Cross-Repo SDK Work
- Plugins run as independent processes managed by the **Plugin Runtime**. The Runtime supports two control transports: `stdio` and `websocket`.
- When LangBot is started directly by a user (not in a container), it spawns and connects to the Runtime over **stdio** (lightweight/personal use).
- When LangBot runs in a container, it connects to a standalone Runtime over **WebSocket** (production).
- The bridge code lives in `src/langbot/pkg/plugin/` (`connector.py`, `handler.py`).
- Relevant config (`data/config.yaml`): `plugin.runtime_ws_url` (e.g. `ws://langbot_plugin_runtime:5400/control/ws`). Start LangBot with `--standalone-runtime` to make it connect to an externally-launched Runtime over WebSocket instead of spawning one over stdio.
When changing SDK contracts used by LangBot:
### Debugging the Plugin Runtime / CLI / SDK
This is documented in detail in the **SDK repo's `AGENTS.md`** and in the wiki page **["调试插件运行时、CLI、SDK" / Plugin Runtime](https://docs.langbot.app/zh/develop/plugin-runtime)**. The short version:
- Clone `LangBot` and `langbot-plugin-sdk` as siblings under one parent dir so the editor resolves shared entities.
- Start a standalone Runtime from the SDK repo: `uv run --no-sync lbp rt` (control port `5400`, debug port `5401`).
- To make LangBot use a locally-modified SDK: from the SDK dir, with LangBot's `.venv` active, run `uv pip install .`, then launch LangBot with `uv run --no-sync main.py --standalone-runtime` (keep `--no-sync` so your local SDK isn't overwritten).
### Debugging the Box (sandbox) runtime
The Box subsystem (`src/langbot/pkg/box/`) is the code sandbox. It picks the first available backend among **Docker / nsjail / E2B**. The standalone Box runtime is launched via the SDK CLI: `lbp box`. Backend selection details, the `lbp box` flags, and the SDK-side architecture are documented in the SDK repo's `AGENTS.md`.
Relevant config (`data/config.yaml`, `box:` section): `box.enabled` (master switch — disabling it also disables the native sandbox tools, skill add/edit, and stdio-mode MCP servers), `box.backend` (`'local'` = Docker/nsjail auto-pick, or `'docker'` / `'nsjail'` / `'e2b'`; also settable via `BOX__BACKEND`), and `box.runtime.endpoint` (external Box runtime base URL, e.g. `ws://127.0.0.1:5410`; empty = local auto-managed runtime). Like the plugin runtime, LangBot can connect to an externally-launched Box runtime by setting that endpoint and starting with `--standalone-box`.
> A common false "No supported sandbox backend (Docker / nsjail / E2B) is available" comes from Docker being installed and running but the current user not being in the `docker` group → `docker info` gets `permission denied` on the socket. Fix: `sudo usermod -aG docker <user>` and restart the backend in a shell that has the new group.
## Development Standards
- LangBot is a global project: **all code comments and docstrings must be in English**, and every user-facing string must support **i18n** (`en_US` + `zh_Hans` at minimum, plus `ja_JP` where the repo already has it).
- LangBot is adopted in both toC and toB scenarios — always consider compatibility and security.
- **Commit message format**: `<type>(<scope>): <subject>`
- `type`: one of `feat`, `fix`, `docs`, `style`, `refactor`, `perf`, `test`, `chore`, etc.
- `scope`: the affected package/module/file/class.
- `subject`: concise description of the change.
### Database migrations (Alembic)
LangBot uses [Alembic](https://alembic.sqlalchemy.org/) for migrations, supporting both SQLite and PostgreSQL from a single set of scripts. Migration files live in `src/langbot/pkg/persistence/alembic/versions/`.
If you change ORM model definitions, generate a migration:
```bash
# from langbot-plugin-sdk, with LangBot's .venv active
uv pip install .
# from LangBot, preserve the locally installed SDK
uv run --no-sync main.py
# Run from the project root (requires data/config.yaml to exist)
uv run python -m langbot.pkg.persistence.alembic_runner autogenerate "description of your change"
```
For standalone runtime debugging:
Review and edit the generated script before committing. Migrations execute automatically on startup. `autogenerate` detects schema changes (add/drop columns, tables, type changes) but **data migrations** (e.g. mutating JSON field contents) must be hand-written into the generated script. `env.py` sets `render_as_batch=True`, so SQLite's ALTER TABLE limits are handled automatically — no need to branch per database. More in the wiki ["开发配置"](https://docs.langbot.app/zh/develop/dev-config#数据库迁移).
```bash
# in langbot-plugin-sdk
uv run --no-sync lbp rt
uv run --no-sync lbp box
When writing a migration, follow these rules:
# in LangBot
uv run --no-sync main.py --standalone-runtime
uv run --no-sync main.py --standalone-box
```
- **Revision id ≤ 32 characters.** PostgreSQL stores `alembic_version.version_num` as `varchar(32)`; a longer id raises `StringDataRightTruncationError` at runtime. Prefer short, descriptive ids like `0005_add_llm_context_length`.
- **Guard every operation against missing tables/columns.** Fresh installs build the schema via `create_all()` and then stamp the Alembic baseline, so a migration may run against a table that already has the change — or, in tests, against an empty database. Check `inspector.get_table_names()` / `inspector.get_columns(...)` before `add_column` / `drop_column`, mirroring the existing migrations.
- **Keep a single linear head.** Chain `down_revision` to the current head; do not create branches. Run the migration tests after adding one: `uv run pytest tests/integration/persistence/ -q` (the PostgreSQL test needs a running PG via `TEST_POSTGRES_URL`).
Config keys to verify in `data/config.yaml` / `src/langbot/templates/config.yaml`:
> **Legacy migration system (deprecated — do not extend).** The old 3.x migration system under `src/langbot/pkg/persistence/migrations/` (`DBMigration` subclasses in `dbmXXX_*.py`, run from `pkg/persistence/mgr.py`) is **frozen**. Do **not** add new `dbmXXX_*.py` files. The chain is capped at `required_database_version = 25` (`pkg/utils/constants.py`); those files only exist to upgrade pre-existing 3.x databases up to the Alembic baseline and are kept read-only. All new schema changes go through Alembic.
- Plugin runtime: `plugin.runtime_ws_url`, default Docker host `langbot_plugin_runtime:5400/control/ws`.
- Box runtime: `box.enabled`, `box.backend`, `box.runtime.endpoint`, Docker host `langbot_box:5410`.
- API/MCP auth: `api.global_api_key`.
## Agent-Facing Surfaces (MCP + Skills)
## Change Rules
LangBot is built to be **agent-friendly**. Three surfaces let AI agents work
with LangBot, and they MUST be kept in lockstep with the HTTP API:
- HTTP API changes that should be agent-accessible must update the matching MCP tool in `src/langbot/pkg/api/mcp/server.py` and the relevant skill under `skills/` in the same pass.
- New schema changes use Alembic under `src/langbot/pkg/persistence/alembic/versions/`; do not add legacy `dbmXXX` migrations.
- New platform behavior belongs in platform adapters only for platform translation; pipeline/business logic belongs in `pkg/pipeline/` or services.
- User-facing strings must support i18n (`en_US`, `zh_Hans`; include `ja_JP` where the repo already does).
- Code comments and docstrings must be English.
- Keep compatibility and security in mind; LangBot is used in both self-hosted/community and toB deployments.
- Commit message format: `<type>(<scope>): <subject>`.
1. **MCP server**`src/langbot/pkg/api/mcp/` exposes a curated subset of the
API as MCP tools at `/mcp` (API-key authenticated, including the
`api.global_api_key` from config.yaml). `server.py` defines the tools (they
call the service layer directly); `mount.py` is the ASGI dispatcher.
2. **In-repo skills**`skills/` is the **single source of truth** for agent
skills (plugin/core/deploy/e2e/MCP-ops). Docs and the landing page link here
rather than embedding their own copies.
3. **API-key auth**`api.global_api_key` (config.yaml) authenticates the API
and MCP without a login session; see `docs/API_KEY_AUTH.md`.
## Runtime Pitfalls
> **Maintenance rule (important).** When you add, remove, or change an HTTP API
> endpoint that should be agent-accessible, you MUST update **both** the matching
> MCP tool in `src/langbot/pkg/api/mcp/server.py` **and** the relevant skill under
> `skills/` (especially `skills/skills/langbot-mcp-ops`). The API, the MCP tool
> surface, and the skills are one system — drift between them is a bug.
- Local stdio Plugin Runtime disconnects do not auto-reconnect; restart LangBot if that path breaks.
- Orphan runtime processes on `5400`/`5401` commonly break plugin debugging.
- Use `uv run --no-sync` after locally installing the SDK, or `uv` may restore the pinned package.
- A false Box “no backend” often means Docker is running but the current user lacks Docker socket permission.
- Do not confuse external MCP servers LangBot connects to (`pkg/provider/tools/loaders/mcp.py`) with LangBot's own `/mcp` server (`pkg/api/mcp/`).
- `CLAUDE.md` is a symlink to this file; edit `AGENTS.md`, not the symlink.
## Principles
## Some Principles
- Keep it simple, stupid.
- Entities should not be multiplied unnecessarily.
-250
View File
@@ -1,250 +0,0 @@
# Architecture
This document is a map of LangBot's moving parts. It is intentionally more stable than a feature guide and more concrete than the README: when you need to change behavior, start here, then follow the file references into the code.
For agent-specific working rules, see `AGENTS.md`. For plugin-runtime and Box-runtime implementation details, also read the sibling SDK repo: [`langbot-plugin-sdk`](https://github.com/langbot-app/langbot-plugin-sdk).
## What LangBot Is
LangBot is an open-source platform for building production IM bots backed by LLMs, agents, RAG, plugins, MCP tools, and a web management panel.
At runtime, one LangBot process owns:
- a Quart/Hypercorn HTTP service and the built web UI on `:5300`;
- messaging-platform adapters such as Discord, Telegram, Slack, WeChat, QQ, WeCom, Lark, DingTalk, KOOK, LINE, Satori, Matrix, and HTTP/WebSocket bots;
- a pipeline engine that turns inbound platform messages into LLM/tool/plugin work and replies;
- persistence, storage, vector database, telemetry, monitoring, and configuration managers;
- bridges to the Plugin Runtime and Box Runtime provided by `langbot-plugin-sdk`;
- an MCP server at `/mcp` exposing a curated agent-facing subset of the service layer.
## Repository Boundary
LangBot is not a single-repo system.
- `LangBot/` is the main product: backend, web UI, platform adapters, pipeline engine, HTTP API, MCP server, RAG, persistence, skills integration, and the bridge code that talks to runtimes.
- `langbot-plugin-sdk/` is published as `langbot-plugin` and pinned in `LangBot/pyproject.toml`. It contains plugin developer APIs, shared entities, `lbp`, the Plugin Runtime (`lbp rt`), and the Box Runtime (`lbp box`).
- Plugins import SDK APIs from `langbot_plugin.*`; the LangBot main process imports the same package for shared entities and runtime protocols.
This split matters. If a change modifies SDK entities, component APIs, action protocols, `lbp rt`, or `lbp box`, verify the sibling SDK repo and install the local SDK into LangBot's virtualenv when testing cross-repo behavior.
## Startup Path
The process entrypoint is small and layered:
1. `main.py` delegates to `langbot.__main__.main()`.
2. `src/langbot/__main__.py` parses `--standalone-runtime`, `--standalone-box`, and `--debug`, checks dependencies, generates missing config/data files, and calls `pkg.core.boot.main()`.
3. `pkg/core/boot.py` executes startup stages in order: `LoadConfigStage`, `GenKeysStage`, `SetupLoggerStage`, `BuildAppStage`, `ShowNotesStage`.
4. `BuildAppStage` constructs the `Application` object by wiring managers, services, runtime connectors, and controllers.
5. `Application.run()` starts the platform manager, query controller, HTTP controller, telemetry/cleanup loops, and plugin initialization.
The central runtime object is `pkg/core/app.py::Application`. It is a service locator for long-lived managers. That is not elegant, but it is the current architectural center; most subsystems receive `ap: Application` and collaborate through it.
## Top-Level Layout
```text
LangBot/
├── main.py # Entrypoint shim
├── pyproject.toml # Python package, deps, pinned langbot-plugin
├── src/langbot/
│ ├── __main__.py # CLI entrypoint and boot handoff
│ ├── pkg/
│ │ ├── core/ # Application, boot stages, task manager
│ │ ├── api/ # HTTP API + MCP server mount
│ │ ├── platform/ # IM adapters and runtime bot manager
│ │ ├── pipeline/ # Message routing and pipeline stages
│ │ ├── provider/ # LLM runners, model manager, tools
│ │ ├── plugin/ # LangBot-side Plugin Runtime connector/handler
│ │ ├── box/ # LangBot-side Box service/connector
│ │ ├── skill/ # Skill metadata/activation integration
│ │ ├── rag/ , vector/ # Knowledge-base and vector DB integration
│ │ ├── persistence/ # SQLAlchemy/SQLModel, Alembic, legacy migrations
│ │ ├── storage/ # Local/S3 file storage abstraction
│ │ └── config/, entity/, utils/, telemetry/, survey/
│ ├── libs/ # Vendored third-party platform SDKs
│ └── templates/ # Default config and component metadata
├── web/ # Vite + React Router + shadcn/ui + Tailwind SPA
├── docker/ # Deployment manifests
├── skills/ # In-repo agent skills, single source of truth
└── tests/ # Unit/integration/e2e/manual tests
```
## The Runtime Graph
The most useful mental model is this graph:
```text
Platform adapter
→ RuntimeBot
→ MessageAggregator
→ QueryPool
→ Controller
→ RuntimePipeline
→ PipelineStage chain
→ RequestRunner / ToolManager / PluginRuntimeConnector / BoxService
→ response via adapter
```
The HTTP and MCP surfaces are parallel entrypoints into the same service layer:
```text
HTTP client / Web UI
→ Quart route group
→ api/http/service/*
→ Application managers / persistence / runtime connectors
MCP client
→ /mcp mount
→ api/mcp/server.py tools
→ the same service layer directly
```
## Message Flow
Inbound platform messages enter through adapter-specific SDK callbacks. The common path is:
1. A platform adapter under `pkg/platform/sources/` converts platform-specific events into SDK message/event entities.
2. `RuntimeBot` in `pkg/platform/botmgr.py` applies pipeline routing rules and either discards the message, pushes it to webhooks, or sends it to the message aggregator.
3. `MessageAggregator` batches/normalizes messages before adding a `Query` to `QueryPool`.
4. `Controller` in `pkg/pipeline/controller.py` selects queries subject to global pipeline concurrency and per-session concurrency.
5. `RuntimePipeline` in `pkg/pipeline/pipelinemgr.py` runs configured pipeline stages using a responsibility-chain style executor that supports generator stages.
6. The chat stage emits plugin events, calls a configured `RequestRunner`, handles streaming/non-streaming responses, records telemetry, and appends conversation history.
7. Output stages send text, cards, chunks, files, or error notices back through the original platform adapter.
Pipeline components are registered by decorators and package import side effects. When adding a new stage, loader, runner, or adapter, check the corresponding preregistration mechanism instead of inventing a second registry.
## Platform Layer
Platform code lives under `pkg/platform/`.
- `botmgr.py` owns runtime bots, routing rules, event logging, webhook pushing, and adapter lifecycle.
- `sources/` contains adapter implementations. Each adapter subclasses `langbot_plugin.api.definition.abstract.platform.adapter.AbstractMessagePlatformAdapter` from the SDK.
- Platform entities such as `MessageChain`, `Image`, `At`, `Voice`, and events come from `langbot-plugin-sdk`, not from this repo.
The platform layer should translate between external platform APIs and LangBot's shared message/event model. It should not contain LLM-provider logic or pipeline business logic.
## Pipeline Layer
Pipeline code lives under `pkg/pipeline/`.
Important pieces:
- `pool.py::QueryPool` stores pending queries and cached in-flight queries for plugin backward-compatible calls.
- `controller.py::Controller` schedules query processing and enforces concurrency.
- `pipelinemgr.py::RuntimePipeline` materializes database pipeline config into a runtime stage chain.
- `process/handlers/chat.py::ChatMessageHandler` is the main LLM conversation handler.
- Stage families include response rules, banned sessions, content filters, preprocessors, rate limits, message truncation, long text handling, response-back, command handling, and wrappers.
Pipelines are configuration-driven. Prefer adding a stage or extending an existing stage family over hard-coding behavior in platform adapters.
## Provider, RAG, and Tools
Provider code lives under `pkg/provider/`.
- `modelmgr/` manages configured model providers and requesters.
- `runners/` implements request runners such as the local agent runner and external workflow integrations.
- `tools/toolmgr.py` aggregates tools from native tools, plugin tools, external MCP servers, and skill-authoring tools.
- `tools/loaders/mcp.py` is the MCP client side: external MCP servers that LangBot connects to for agent tools.
- RAG lives across `pkg/rag/`, `pkg/vector/`, model services, and plugin KnowledgeEngine actions.
Do not confuse LangBot's MCP client side with LangBot's own MCP server at `/mcp`; they are different surfaces.
## Plugin System
The plugin system crosses the repo boundary.
In this repo:
- `pkg/plugin/connector.py` connects LangBot to the Plugin Runtime over stdio or WebSocket.
- `pkg/plugin/handler.py` exposes LangBot actions to the runtime and calls runtime actions for plugin operations.
- `pkg/provider/tools/loaders/plugin.py` exposes plugin Tool components to LLM runners.
- Pipeline handlers emit SDK events such as normal-message events and prompt-processing events.
In `langbot-plugin-sdk`:
- `src/langbot_plugin/api/` defines `BasePlugin`, component base classes, message/event entities, contexts, proxies, and manifests.
- `src/langbot_plugin/runtime/` implements `lbp rt`, plugin discovery, dependency installation, process launching, and control/debug connections.
- `src/langbot_plugin/entities/io/` defines the action protocol shared by LangBot, runtime, and plugin processes.
The Plugin Runtime supports stdio and WebSocket control transports. Direct local LangBot runs usually spawn the runtime over stdio. Containerized/standalone deployments connect over WebSocket using `plugin.runtime_ws_url` and `--standalone-runtime`.
## Box Runtime and Skills
Box is the sandbox subsystem used by native agent tools, stdio MCP servers, skill authoring, and managed processes.
In this repo:
- `pkg/box/service.py` is the application-facing facade for exec, sessions, managed processes, skill CRUD, status, reconnects, quotas, mounts, and sandbox profiles.
- `pkg/box/connector.py` connects to the Box Runtime over stdio, Windows subprocess+WebSocket, or remote WebSocket.
- `pkg/provider/tools/loaders/native.py`, `mcp_stdio.py`, and skill loaders depend on Box availability.
- `pkg/skill/manager.py` loads skills from the Box runtime, falling back to local `data/skills` when needed.
In `langbot-plugin-sdk`:
- `src/langbot_plugin/box/server.py` implements `lbp box` and the WebSocket endpoints on `:5410`.
- `src/langbot_plugin/box/runtime.py` owns sandbox sessions and managed processes.
- `backend.py`, `nsjail_backend.py`, and `e2b_backend.py` implement sandbox backends.
- `skill_store.py` manages skill packages from the Box side.
Important config keys live under `box:` in `src/langbot/templates/config.yaml`: `box.enabled`, `box.backend`, `box.runtime.endpoint`, and `box.local.*`. Start LangBot with `--standalone-box` when connecting to an externally launched Box runtime.
## HTTP API, Web UI, and MCP Server
`pkg/api/http/controller/main.py` builds a Quart app, registers route groups, serves the built SPA, and wraps the ASGI app with the MCP dispatcher.
- HTTP route groups live under `pkg/api/http/controller/groups/`.
- Service-layer logic lives under `pkg/api/http/service/`.
- The built web UI is served from the frontend build path with SPA fallback.
- The MCP server lives under `pkg/api/mcp/` and is mounted at `/mcp`.
The MCP server intentionally exposes a curated subset of the API. Tools call service classes directly rather than making HTTP requests back into LangBot.
Maintenance rule: when adding, removing, or changing an HTTP endpoint that should be agent-accessible, update the matching MCP tool and the relevant in-repo skill under `skills/` in the same pass.
## Persistence and Configuration
Persistence is centered on `pkg/persistence/mgr.py`.
- SQLite is the default database; PostgreSQL is supported.
- Models live under `pkg/entity/persistence/`.
- Fresh schemas are created from metadata, then legacy migrations run up to the frozen 3.x baseline, then Alembic migrations run to head.
- New schema changes should use Alembic under `pkg/persistence/alembic/versions/`; do not extend the frozen legacy migration chain.
Configuration starts from `src/langbot/templates/config.yaml` and is generated into `data/config.yaml` on first run. Most long-lived managers read from `ap.instance_config.data`.
## Frontend
The frontend lives in `web/` and is a Vite SPA using React Router 7, shadcn/ui, Tailwind CSS, and pnpm. It is not Next.js, despite some historical filenames.
In development, `pnpm dev` serves the UI on `:3000` and reads `VITE_API_BASE_URL` to call the backend on `:5300`. In production, the built frontend is packaged into the Python distribution and served by the backend.
Keep frontend API behavior aligned with `pkg/api/http/service/` and route groups. User-facing strings must go through the existing i18n setup.
## Agent-Facing Surfaces
LangBot is deliberately agent-friendly. The agent-facing surfaces are part of the architecture, not extra docs.
- `skills/` is the single source of truth for in-repo skills.
- `pkg/api/mcp/server.py` exposes the LangBot MCP server at `/mcp`.
- `api.global_api_key` authenticates API/MCP access without a browser login.
- `AGENTS.md` and `ARCHITECTURE.md` tell coding agents how the repo works.
When one of these changes, update the others if the behavior or contract changed. API, MCP tools, and skills are one system; drift is a bug.
## Where to Change Things
- New HTTP API: add/adjust a service in `pkg/api/http/service/`, a route group in `pkg/api/http/controller/groups/`, tests, and MCP/skills if agent-accessible.
- New platform adapter: add a `pkg/platform/sources/*` adapter, component metadata/templates as needed, i18n, docs, and tests/smoke coverage.
- New pipeline behavior: add or extend a pipeline stage family under `pkg/pipeline/`; avoid putting pipeline rules in adapters.
- New LLM provider/requester: work under `pkg/provider/modelmgr/` and related service/UI surfaces.
- New LLM tool source: extend `pkg/provider/tools/loaders/` and `ToolManager` intentionally.
- New plugin component/API/protocol: change `langbot-plugin-sdk` first or in lockstep, then update LangBot bridge code.
- New Box capability: change both `pkg/box/` and `langbot-plugin-sdk/src/langbot_plugin/box/`, plus config and tests.
- New database schema: add an Alembic migration, not a legacy `dbmXXX` migration.
## Design Biases
- Keep platform translation, pipeline orchestration, provider execution, and runtime protocols separate.
- Reuse existing registries and service layers instead of adding parallel paths.
- Prefer small, explicit agent surfaces over exposing every internal API.
- Treat cross-repo contracts with the SDK as public interfaces.
- Test behavior at the narrowest useful layer first, then add integration/e2e coverage for runtime or platform changes.
+2 -2
View File
@@ -5,7 +5,7 @@
<div align="center">
<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
<h3>Production-grade platform for building agentic IM bots.</h3>
<h4>Quickly build, debug, and ship AI bots to Slack, Discord, Telegram, WeChat, and more.</h4>
@@ -136,7 +136,7 @@ docker compose --profile all up -d
| [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | GPU Platform | ✅ |
| [ShengSuanYun](https://www.shengsuanyun.com/?from=CH_KYIPP758) | GPU Platform | ✅ |
| [接口 AI](https://jiekou.ai/) | Gateway | ✅ |
| [302.AI](https://share.302ai.cn/SuTG99) | Gateway | ✅ |
| [302.AI](https://share.302.ai/SuTG99) | Gateway | ✅ |
| [Qiniu](https://www.qiniu.com/ai/agent) | Gateway | ✅ |
[→ View all integrations](https://link.langbot.app/en/docs/features)
+1 -1
View File
@@ -136,7 +136,7 @@ docker compose --profile all up -d
| [优云智算](https://www.compshare.cn/?ytag=GPU_YY-gh_langbot) | GPU 平台 | ✅ |
| [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | GPU 平台 | ✅ |
| [接口 AI](https://jiekou.ai/) | 聚合平台 | ✅ |
| [302.AI](https://share.302ai.cn/SuTG99) | 聚合平台 | ✅ |
| [302.AI](https://share.302.ai/SuTG99) | 聚合平台 | ✅ |
| [小马算力](https://www.tokenpony.cn/453z1) | 聚合平台 | ✅ |
| [百宝箱Tbox](https://www.tbox.cn/open) | 智能体平台 | ✅ |
| [七牛云Qiniu](https://www.qiniu.com/ai/agent) | 聚合平台 | ✅ |
+2 -2
View File
@@ -5,7 +5,7 @@
<div align="center">
<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
<h3>Plataforma de grado de producción para construir bots de mensajería instantánea con agentes de IA.</h3>
<h4>Construya, depure y despliegue bots de IA rápidamente en Slack, Discord, Telegram, WeChat y más.</h4>
@@ -135,7 +135,7 @@ docker compose --profile all up -d
| [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | Plataforma GPU | ✅ |
| [ShengSuanYun](https://www.shengsuanyun.com/?from=CH_KYIPP758) | Plataforma GPU | ✅ |
| [接口 AI](https://jiekou.ai/) | Pasarela | ✅ |
| [302.AI](https://share.302ai.cn/SuTG99) | Pasarela | ✅ |
| [302.AI](https://share.302.ai/SuTG99) | Pasarela | ✅ |
| [Qiniu](https://www.qiniu.com/ai/agent) | Pasarela | ✅ |
[→ Ver todas las integraciones](https://link.langbot.app/en/docs/features)
+2 -2
View File
@@ -5,7 +5,7 @@
<div align="center">
<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
<h3>Plateforme de niveau production pour construire des bots de messagerie instantanée avec agents IA.</h3>
<h4>Créez, déboguez et déployez rapidement des bots IA sur Slack, Discord, Telegram, WeChat et plus.</h4>
@@ -132,7 +132,7 @@ docker compose --profile all up -d
| [ModelScope](https://modelscope.cn/docs/model-service/API-Inference/intro) | Passerelle | ✅ |
| [GiteeAI](https://ai.gitee.com/) | Passerelle | ✅ |
| [接口 AI](https://jiekou.ai/) | Passerelle | ✅ |
| [302.AI](https://share.302ai.cn/SuTG99) | Passerelle | ✅ |
| [302.AI](https://share.302.ai/SuTG99) | Passerelle | ✅ |
| [CompShare](https://www.compshare.cn/?ytag=GPU_YY-gh_langbot) | Plateforme GPU | ✅ |
| [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | Plateforme GPU | ✅ |
| [ShengSuanYun](https://www.shengsuanyun.com/?from=CH_KYIPP758) | Plateforme GPU | ✅ |
+2 -2
View File
@@ -5,7 +5,7 @@
<div align="center">
<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
<h3>AIエージェント搭載IMボットを構築するための本番グレードプラットフォーム。</h3>
<h4>Slack、Discord、Telegram、WeChat などに AI ボットを素早く構築、デバッグ、デプロイ。</h4>
@@ -135,7 +135,7 @@ docker compose --profile all up -d
| [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | GPUプラットフォーム | ✅ |
| [ShengSuanYun](https://www.shengsuanyun.com/?from=CH_KYIPP758) | GPUプラットフォーム | ✅ |
| [接口 AI](https://jiekou.ai/) | ゲートウェイ | ✅ |
| [302.AI](https://share.302ai.cn/SuTG99) | ゲートウェイ | ✅ |
| [302.AI](https://share.302.ai/SuTG99) | ゲートウェイ | ✅ |
| [Qiniu](https://www.qiniu.com/ai/agent) | ゲートウェイ | ✅ |
[→ すべての統合を表示](https://link.langbot.app/en/docs/features)
+2 -2
View File
@@ -5,7 +5,7 @@
<div align="center">
<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
<h3>AI 에이전트 IM 봇 구축을 위한 프로덕션 등급 플랫폼.</h3>
<h4>Slack, Discord, Telegram, WeChat 등에 AI 봇을 빠르게 구축, 디버그 및 배포.</h4>
@@ -135,7 +135,7 @@ docker compose --profile all up -d
| [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | GPU 플랫폼 | ✅ |
| [ShengSuanYun](https://www.shengsuanyun.com/?from=CH_KYIPP758) | GPU 플랫폼 | ✅ |
| [接口 AI](https://jiekou.ai/) | 게이트웨이 | ✅ |
| [302.AI](https://share.302ai.cn/SuTG99) | 게이트웨이 | ✅ |
| [302.AI](https://share.302.ai/SuTG99) | 게이트웨이 | ✅ |
| [Qiniu](https://www.qiniu.com/ai/agent) | 게이트웨이 | ✅ |
[→ 모든 통합 보기](https://link.langbot.app/en/docs/features)
+2 -2
View File
@@ -5,7 +5,7 @@
<div align="center">
<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
<h3>Платформа производственного уровня для создания агентных IM-ботов.</h3>
<h4>Быстро создавайте, отлаживайте и развертывайте ИИ-ботов в Slack, Discord, Telegram, WeChat и других платформах.</h4>
@@ -131,7 +131,7 @@ docker compose --profile all up -d
| [Volc Engine Ark](https://console.volcengine.com/ark/region:ark+cn-beijing/model?vendor=Bytedance&view=LIST_VIEW) | Шлюз | ✅ |
| [ModelScope](https://modelscope.cn/docs/model-service/API-Inference/intro) | Шлюз | ✅ |
| [GiteeAI](https://ai.gitee.com/) | Шлюз | ✅ |
| [302.AI](https://share.302ai.cn/SuTG99) | Шлюз | ✅ |
| [302.AI](https://share.302.ai/SuTG99) | Шлюз | ✅ |
| [接口 AI](https://jiekou.ai/) | Шлюз | ✅ |
| [CompShare](https://www.compshare.cn/?ytag=GPU_YY-gh_langbot) | Платформа GPU | ✅ |
| [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | Платформа GPU | ✅ |
+1 -1
View File
@@ -137,7 +137,7 @@ docker compose --profile all up -d
| [優雲智算](https://www.compshare.cn/?ytag=GPU_YY-gh_langbot) | GPU 平台 | ✅ |
| [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | GPU 平台 | ✅ |
| [接口 AI](https://jiekou.ai/) | 聚合平台 | ✅ |
| [302.AI](https://share.302ai.cn/SuTG99) | 聚合平台 | ✅ |
| [302.AI](https://share.302.ai/SuTG99) | 聚合平台 | ✅ |
| [Qiniu](https://www.qiniu.com/ai/agent) | 聚合平台 | ✅ |
### TTS(語音合成)
+2 -2
View File
@@ -5,7 +5,7 @@
<div align="center">
<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
<h3>Nền tảng cấp sản xuất để xây dựng bot IM với AI agent.</h3>
<h4>Xây dựng, gỡ lỗi và triển khai bot AI nhanh chóng trên Slack, Discord, Telegram, WeChat và nhiều nền tảng khác.</h4>
@@ -135,7 +135,7 @@ docker compose --profile all up -d
| [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | Nền tảng GPU | ✅ |
| [ShengSuanYun](https://www.shengsuanyun.com/?from=CH_KYIPP758) | Nền tảng GPU | ✅ |
| [接口 AI](https://jiekou.ai/) | Cổng | ✅ |
| [302.AI](https://share.302ai.cn/SuTG99) | Cổng | ✅ |
| [302.AI](https://share.302.ai/SuTG99) | Cổng | ✅ |
| [Qiniu](https://www.qiniu.com/ai/agent) | Cổng | ✅ |
[→ Xem tất cả tích hợp](https://link.langbot.app/en/docs/features)
+1 -2
View File
@@ -62,12 +62,11 @@ services:
- TZ=Asia/Shanghai
# Unified env-override convention: SECTION__SUBSECTION__KEY overrides the
# matching config.yaml field (see LoadConfigStage). These map onto
# box.* and are forwarded to the Box runtime via INIT RPC.
# box.local.* and are forwarded to the Box runtime via INIT RPC.
- BOX__LOCAL__HOST_ROOT=${LANGBOT_BOX_ROOT:-${PWD}/data/box}
- BOX__LOCAL__DEFAULT_WORKSPACE=default
- BOX__LOCAL__SKILLS_ROOT=skills
- BOX__LOCAL__ALLOWED_MOUNT_ROOTS=${LANGBOT_BOX_ROOT:-${PWD}/data/box}
- BOX__DOCKER__CPU_LIMIT_ENABLED=${LANGBOT_BOX_DOCKER_CPU_LIMIT_ENABLED:-true}
ports:
- 5300:5300 # For web ui and webhook callback
- 2280-2285:2280-2285 # For platform reverse connection
-575
View File
@@ -1,575 +0,0 @@
# HTTP Bot Adapter — Design Document
> Status: **Implemented** · Branch: `feat/http-bot-adapter` · Author: LangBot core
>
> A first-class, **standalone** message-platform adapter (`http_bot`) that lets
> any external system (e.g. LangBot Space ticketing, an internal back-office, a
> CRM, a custom web app) talk to a LangBot pipeline over plain HTTP — **inbound**
> by POSTing messages in, **outbound** by receiving replies on a callback URL —
> with full support for the pipeline's native N→1 aggregation and 1→M
> multi-reply semantics, and **without** holding a long-lived WebSocket
> connection.
>
> **Shipped in this branch:**
> - `src/langbot/pkg/platform/sources/http_bot.yaml` — adapter manifest (auto-discovered)
> - `src/langbot/pkg/platform/sources/http_bot.py` — `HttpBotAdapter`
> - `src/langbot/pkg/platform/sources/http_bot_signing.py` — HMAC helpers
> - `src/langbot/pkg/platform/sources/http_bot.svg` — icon
> - `docs/platforms/http-bot.md` — integration guide
> - `docs/http-bot-openapi.json` — machine-readable contract
> - `examples/http-bot/` — Python + TypeScript reference clients
>
> **Final decisions (resolving the original open questions):**
> 1. Callback URL is **config-only** — never accepted per-message (SSRF closed).
> 2. **Session reset is provided** — `POST /bots/<uuid>/reset` keyed by `session_id`.
> 3. Reference **clients are provided** — `examples/http-bot/client.py` + `client.ts`.
> 4. **Sync convenience mode is included** — `POST /bots/<uuid>/sync` (opt-in, lossy).
---
## 1. Background & Motivation
### 1.1 The concrete need
LangBot Space wants to use a LangBot pipeline as the brain for **ticket
handling**. The integration is **server-to-server**: Space's backend pushes a
user's ticket messages into LangBot and renders LangBot's replies back into the
ticket thread.
This interaction is **not** request/response shaped:
- **N → 1**: a user may fire several messages in a row ("the app crashed" …
"when I click export" … "here's a screenshot"). The pipeline's
**message aggregation** feature should debounce and merge these into one turn.
- **1 → N**: a single turn may yield **multiple** outbound messages — a tool/
function call narrating progress, a plugin emitting several cards, a streamed
answer split into chunks.
### 1.2 Why the existing options don't fit
LangBot today exposes exactly one externally-reachable way to drive a pipeline
that is **not** tied to a specific IM vendor: the **WebSocket** path
(`/api/v1/pipelines/<uuid>/ws/connect` for dashboard debug, and
`/api/v1/embed/<bot_uuid>/ws/connect` for the embeddable web widget).
For a server-to-server integration the WebSocket path has real friction:
| Problem | Detail |
|---|---|
| Long-lived connection | Caller must maintain a socket, heartbeats, and reconnect logic for what is fundamentally a fire-and-collect workload. |
| Session identity | Inbound messages are keyed by the transient `connection_id` (`websocket_{connection_id}`); the caller **cannot supply a stable, business-meaningful session id** (e.g. a ticket number). Multi-ticket isolation is not expressible. |
| Auth mismatch | The debug socket is gated by the **dashboard JWT** (must not be handed to an external service); the embed socket is gated by **Cloudflare Turnstile** (a *browser* human-check that a backend cannot satisfy). Neither is a server-to-server credential. |
| In-memory, single-process state | Session history lives in process memory and is lost on restart. |
> **Key realisation.** The N→1 / 1→M behaviour the caller wants is **not**
> provided by WebSocket — it is provided by the **pipeline** (aggregation +
> the adapter being free to call `reply_message` any number of times). It is
> therefore **transport-independent**. We can deliver the exact same semantics
> over a far lighter HTTP transport.
### 1.3 Why a *new, standalone* adapter (not a refactor of an existing one)
The brief is explicit: **do not reuse / fork an existing vendor adapter.** The
vendor adapters (`lark`, `wecom`, `qqofficial`, `slack`, …) carry vendor-specific
signature schemes, payload shapes, and message-segment mappings. Bending one of
them into a "generic" mode would couple a public integration surface to one
vendor's quirks and make the developer experience worse for everyone.
Instead we ship `http_bot` as a clean, independent adapter whose **entire
contract is LangBot's own** — documented, versioned, and designed front-to-back
around *integrator* developer experience.
---
## 2. Goals & Non-Goals
### Goals
- **G1** A standalone `http_bot` adapter, selectable like any other platform
adapter in the dashboard, with its own config schema and docs.
- **G2** **Inbound**: external systems POST messages to a stable LangBot URL,
carrying a **caller-defined `session_id`** that maps 1:1 to a LangBot session.
- **G3** **Outbound**: LangBot delivers each reply by POSTing to a
caller-configured **callback URL**; one turn may produce **many** callbacks.
- **G4** Preserve pipeline-native **N→1 aggregation** and **1→M multi-reply**.
- **G5** Server-to-server **auth**: shared-secret HMAC request signing both
directions (no JWT, no Turnstile, no long-lived socket).
- **G6** **Great DX**: copy-pasteable curl, a tiny reference client, an OpenAPI
fragment, idempotency, clear error envelope, and a local echo-server recipe.
### Non-Goals
- Not replacing or deprecating the WebSocket / embed widget path (that remains
the right tool for *browser*, real-time, streaming chat UIs).
- Not a synchronous "one request → one response" RPC (explicitly rejected: it
cannot express 1→M; see §9 for the optional sync convenience mode).
- No built-in message **persistence/replay** in v1 (callbacks are at-least-once
best-effort; durability is the caller's responsibility — see §8).
- No multi-tenant API-key management UI in v1 (one secret per bot; see §11).
---
## 3. How LangBot routes a message (the parts we plug into)
Understanding the existing flow is what makes this adapter cheap. A message
flows through these stages (verified against current `master`):
```
INBOUND OUTBOUND
external POST ─┐ ┌─ reply_message()
▼ │ reply_message_chunk()
POST /bots/<bot_uuid> (unified webhook router, AuthType.NONE)
│ webhooks.py → adapter.handle_unified_webhook(bot_uuid, path, request)
▼ │
HttpBotAdapter.handle_unified_webhook │ (called 0..N times
• verify HMAC signature │ per turn by the
• parse {session_id, message[]} │ pipeline / plugins)
• build FriendMessage / GroupMessage │
• fire registered listener ───────────────┐ │
│ │ │
▼ ▼ │
botmgr.on_friend_message / on_group_message │
• (optional) webhook_pusher fan-out │
• msg_aggregator.add_message(...) ── N→1 debounce ──►│
│ │
▼ │
query_pool → pipeline.run() ─── invokes adapter ─────┘
reply methods 1..M times
```
Two framework facts we rely on:
1. **N→1 aggregation is free.** `botmgr` hands every inbound event to
`self.ap.msg_aggregator.add_message(...)`, which debounces per
`session_id` and merges consecutive messages into one pipeline turn
(`pkg/pipeline/aggregator.py`). The adapter does nothing special.
2. **1→M is free.** The pipeline (and any plugin in the chain) calls
`adapter.reply_message()` / `reply_message_chunk()` **as many times as it
wants** per turn. The adapter's only job is to deliver each call outward.
For `http_bot` that means: **one outbound callback POST per call.**
3. **A unified inbound route already exists.** `WebhookRouterGroup`
(`pkg/api/http/controller/groups/webhooks.py`) maps
`POST /bots/<bot_uuid>[/<path>]` (auth `NONE`) to
`adapter.handle_unified_webhook(bot_uuid, path, request)`. `http_bot`
implements that method and is reachable **without registering any new
route** — it does its own signature verification, exactly like the vendor
webhook adapters do.
> Net new code is essentially: one `http_bot.py` adapter, one `http_bot.yaml`
> schema, signing helpers, and docs. No router, aggregator, or pipeline changes.
---
## 4. Architecture Overview
```
┌────────────────────┐ (1) inbound: POST signed message
│ External system │ ──────────────────────────────────────────────► ┌──────────────────────┐
│ (LangBot Space, │ POST /bots/<bot_uuid> │ LangBot │
│ CRM, web app …) │ X-LB-Signature, X-LB-Timestamp │ │
│ │ { session_id, message:[...] } │ HttpBotAdapter │
│ - callback server │ ◄────────────────────────────────────────────── │ (platform/sources) │
│ (receives │ (4) outbound: POST signed reply(s) │ │
│ replies) │ POST <callback_url> │ pipeline + aggregator│
└────────────────────┘ X-LB-Signature, X-LB-Timestamp └──────────────────────┘
{ session_id, sequence, is_final,
message:[...] } (sent 1..M times)
```
- The adapter is **stateless across requests** at the HTTP layer; session
continuity is carried by `session_id` and resolved by LangBot's normal
session manager.
- **Inbound** and **outbound** are **independent HTTP exchanges**. LangBot does
not answer the inbound POST with the pipeline result; it `202 Accepts` it and
later POSTs the reply(s) to the callback URL. This is what makes 1→M natural.
---
## 5. Configuration Schema (`http_bot.yaml`)
Follows the existing `MessagePlatformAdapter` manifest convention (cf.
`slack.yaml`). Fields:
| field | type | required | purpose |
|---|---|---|---|
| `inbound_secret` | string (secret) | yes | HMAC key the **caller** uses to sign inbound POSTs; LangBot verifies. |
| `callback_url` | string (url) | no* | Where LangBot POSTs replies. *Optional if the caller supplies `callback_url` per-message (see §6.1); a static default lives here. |
| `outbound_secret` | string (secret) | no | HMAC key LangBot uses to sign outbound callbacks; caller verifies. Defaults to `inbound_secret` if empty. |
| `default_session_type` | enum `person`/`group` | no | Default when a message omits `session_type`. Default `person`. |
| `signature_required` | bool | no | If `false`, skip inbound signature check (dev only; logs a warning). Default `true`. |
| `callback_timeout` | int (seconds) | no | Per-callback HTTP timeout. Default `15`. |
| `callback_max_retries` | int | no | Retries on 5xx/timeout with backoff. Default `3`. |
| `webhook_url` | webhook-url (display) | — | Read-only field rendering the inbound URL `…/bots/<bot_uuid>` for copy-paste, like other webhook adapters. |
Manifest sketch (i18n labels elided for brevity):
```yaml
apiVersion: v1
kind: MessagePlatformAdapter
metadata:
name: http_bot
label: { en_US: "HTTP Bot", zh_Hans: "HTTP 通用接入" }
description:
en_US: "Integrate any backend over plain HTTP. Push messages in, receive replies on a callback URL. Server-to-server, no long-lived connection."
zh_Hans: "通过 HTTP 接入任意后端系统。推入消息、在回调地址接收回复。面向服务间集成,无需长连接。"
icon: http_bot.svg
spec:
categories: [popular, global]
help_links:
zh: https://docs.langbot.app/zh/platforms/http-bot
en: https://docs.langbot.app/en/platforms/http-bot
config:
- { name: inbound_secret, type: string, required: true, default: "" }
- { name: callback_url, type: string, required: false, default: "" }
- { name: outbound_secret, type: string, required: false, default: "" }
- { name: default_session_type, type: select, required: false, default: "person",
options: [person, group] }
- { name: signature_required, type: boolean, required: false, default: true }
- { name: callback_timeout, type: integer, required: false, default: 15 }
- { name: callback_max_retries, type: integer, required: false, default: 3 }
- { name: webhook_url, type: webhook-url, required: false, default: "" }
execution:
python:
path: ./http_bot.py
attr: HttpBotAdapter
```
---
## 6. The HTTP Contract (this is the DX surface)
### 6.1 Inbound — push a message into LangBot
```
POST /bots/{bot_uuid}
Content-Type: application/json
X-LB-Timestamp: 1718000000
X-LB-Signature: sha256=<hex hmac>
X-LB-Idempotency-Key: <uuid> # optional, dedup window
```
Body:
```jsonc
{
"session_id": "ticket-10293", // REQUIRED. Caller-defined. Maps 1:1 to a LangBot session.
"session_type": "person", // optional, "person" | "group"; default from config
"sender": { // optional metadata, surfaced to pipeline/plugins
"id": "user-5567",
"name": "Alice"
},
"message": [ // REQUIRED. A LangBot MessageChain (list of segments).
{ "type": "Plain", "text": "Export keeps failing on the dashboard." },
{ "type": "Image", "url": "https://.../screenshot.png" }
]
}
```
Response (LangBot does **not** block on the pipeline):
```jsonc
// 202 Accepted
{
"code": 0,
"msg": "accepted",
"data": {
"session_id": "ticket-10293",
"accepted_message_id": "in_01H....", // server-assigned id for this inbound message
"aggregating": true // true if buffered by the aggregator
}
}
```
**N→1 in practice.** Fire three POSTs with the same `session_id` inside the
aggregation window → the pipeline runs **once** with the three messages merged.
No special flag needed; this is the aggregator's default behaviour when enabled
on the pipeline.
### 6.2 Outbound — LangBot delivers replies to your callback
For each `reply_message` / `reply_message_chunk` the pipeline emits, LangBot
POSTs to `callback_url`:
```
POST {callback_url}
Content-Type: application/json
X-LB-Timestamp: 1718000001
X-LB-Signature: sha256=<hex hmac over body>
```
Body:
```jsonc
{
"session_id": "ticket-10293", // echoes the inbound session
"reply_to": "in_01H....", // the inbound message id this answers
"sequence": 1, // 1-based ordinal within this turn (for 1→M ordering)
"is_final": false, // false for intermediate/streamed parts
"stream": false, // true when this is a streamed chunk
"message": [
{ "type": "Plain", "text": "Looking into it — checking your export logs…" }
],
"timestamp": "2026-06-22T09:00:01Z"
}
```
**1→M in practice.** A turn that fires a function call then a final answer
produces e.g.:
```
POST callback → { sequence: 1, is_final: false, message: ["Checking logs…"] }
POST callback → { sequence: 2, is_final: false, message: ["Found 2 failed exports."] }
POST callback → { sequence: 3, is_final: true, message: ["Fixed. Try again now."] }
```
The caller stitches by `session_id` + `sequence`, and knows the turn is complete
when `is_final: true` arrives.
Your callback endpoint should return `200` quickly. A non-2xx triggers retry
with backoff (`callback_max_retries`).
### 6.3 Error envelope (inbound)
Consistent, machine-readable; never leak internals:
```jsonc
{ "code": 40101, "msg": "invalid signature", "data": null }
```
| HTTP | code | meaning |
|---|---|---|
| 202 | 0 | accepted |
| 400 | 40001 | malformed body / missing `session_id` or `message` |
| 401 | 40101 | bad/expired signature |
| 403 | 40301 | bot disabled |
| 404 | 40401 | bot_uuid not found / not an `http_bot` adapter |
| 409 | 40901 | duplicate idempotency key (already accepted) |
| 413 | 41301 | message too large |
| 500 | 50001 | internal error |
---
## 7. Signing scheme (both directions)
Symmetric, dependency-free HMAC-SHA256 — trivial to implement in any language.
```
signing_string = "{timestamp}.{raw_request_body}"
signature = "sha256=" + hex(HMAC_SHA256(secret, signing_string))
```
Verification rules:
- Reject if `|now - timestamp| > 300s` (replay window).
- Constant-time compare (`hmac.compare_digest`).
- Inbound verified with `inbound_secret`; outbound signed with
`outbound_secret` (falls back to `inbound_secret`).
- `signature_required: false` bypasses verification **and logs a warning**
intended only for local development behind a trusted network.
Reference (Python, ~6 lines):
```python
import hmac, hashlib, time
def sign(secret: str, body: bytes, ts: int | None = None) -> tuple[str, str]:
ts = ts or int(time.time())
mac = hmac.new(secret.encode(), f"{ts}.".encode() + body, hashlib.sha256)
return str(ts), "sha256=" + mac.hexdigest()
```
---
## 8. Delivery semantics & reliability
- **Inbound**: `202 Accepted` means *queued*, not *processed*. Use
`X-LB-Idempotency-Key` to make client retries safe (dedup window, e.g. 10 min).
- **Outbound**: **at-least-once**, best-effort. Retries on timeout/5xx with
exponential backoff up to `callback_max_retries`. Callbacks for one
`session_id` are delivered **in `sequence` order** (serialised per session);
across sessions they may interleave.
- **No persistence in v1**: if LangBot restarts mid-turn, in-flight callbacks
may be lost. Durable replay is deferred (see §13). Callers needing exactly-once
should dedup on `(session_id, reply_to, sequence)`.
- **Backpressure**: the adapter must not block the pipeline on slow callbacks —
outbound POSTs run on a per-session ordered queue with the configured timeout.
---
## 9. Optional: synchronous convenience mode (v1.1, behind a flag)
Some simple callers genuinely want "POST a message, get the reply in the HTTP
response" and don't care about streaming/multi-part. We can offer an **opt-in**
sync endpoint that internally waits for `is_final` and **collapses** all 1→M
parts into one array:
```
POST /bots/{bot_uuid}/sync → 200 { session_id, message: [ ...all parts concatenated... ] }
```
Implemented by attaching a per-request future that resolves on the final reply,
with a hard timeout. This is a **convenience wrapper** over the same machinery,
explicitly documented as lossy for streaming/ordering. Not in v1 core.
---
## 10. Adapter implementation sketch (`platform/sources/http_bot.py`)
Implements `AbstractMessagePlatformAdapter`. Key methods:
```python
class HttpBotAdapter(AbstractMessagePlatformAdapter):
listeners: dict = pydantic.Field(default_factory=dict, exclude=True)
# --- inbound -------------------------------------------------------
async def handle_unified_webhook(self, bot_uuid, path, request):
body = await request.get_body()
if self.config.get("signature_required", True):
if not self._verify(request, body):
return jsonify({"code": 40101, "msg": "invalid signature"}), 401
data = json.loads(body)
session_id = data["session_id"] # caller-defined identity
session_type = data.get("session_type", self.config.get("default_session_type", "person"))
chain = MessageChain.model_validate(data["message"])
event = self._build_event(session_type, session_id, data.get("sender"), chain)
# remember where to send replies for this session
self._callback_for[session_id] = data.get("callback_url") or self.config.get("callback_url")
# fire the registered listener → botmgr → msg_aggregator (N→1) → pipeline
if type(event) in self.listeners:
asyncio.create_task(self.listeners[type(event)](event, self))
return jsonify({"code": 0, "msg": "accepted",
"data": {"session_id": session_id, "accepted_message_id": event.message_id}}), 202
# --- outbound (called 1..M times per turn by the pipeline) ---------
async def reply_message(self, message_source, message, quote_origin=False):
return await self._post_callback(message_source, message, is_final=True, stream=False)
async def reply_message_chunk(self, message_source, bot_message, message,
quote_origin=False, is_final=False):
return await self._post_callback(message_source, message, is_final=is_final, stream=True)
async def is_stream_output_supported(self) -> bool:
return True
def register_listener(self, event_type, func): self.listeners[event_type] = func
def unregister_listener(self, event_type, func): self.listeners.pop(event_type, None)
async def run_async(self): pass # nothing to poll; purely webhook-driven
async def kill(self): pass
```
`_post_callback` resolves the session's callback URL, assigns the next
`sequence`, signs the body, and enqueues an ordered, retrying POST.
Session→callback mapping is kept in a small in-memory dict keyed by
`session_id` (acceptable for v1; a turn's callback URL is captured at inbound
time so replies always have a destination even if config later changes).
---
## 11. Security considerations
- **Inbound route is `AuthType.NONE`** at the framework level (same as all
webhook adapters) — the adapter **must** enforce HMAC itself. Default
`signature_required: true`.
- **Timestamp window** (±300s) + idempotency key blunt replay.
- **SSRF on callback_url**: validate scheme (`https` in prod), and consider an
allow-list / block of private CIDRs since LangBot initiates the POST. Document
this; enforce in code where feasible.
- **Secret storage**: secrets live in the bot's `adapter_config` like every
other adapter credential; surfaced as `type: string`/secret in the dashboard.
- **One secret per bot** in v1. Per-caller key rotation / multiple keys is a
future enhancement (§13).
---
## 12. Developer Experience (explicit deliverables)
The whole point of a standalone adapter is that **integrating is pleasant**. v1
ships:
1. **`docs/platforms/http-bot.md`** — task-oriented integration guide:
create the bot → copy inbound URL → set secret → stand up a callback
endpoint → send first message → handle 1→M.
2. **Copy-paste curl** for the first message (with a working signing one-liner).
3. **Reference clients** (≤50 LOC each) in `examples/http-bot/`:
`client.py` (push + a Flask/Quart callback receiver) and `client.ts`.
4. **OpenAPI fragment** `docs/http-bot-openapi.json` describing inbound +
callback shapes, so integrators can codegen.
5. **Local echo recipe**: a one-command callback server that prints every
reply, so a developer sees N→1 and 1→M working in under five minutes.
6. **Postman/Hoppscotch collection** (nice-to-have).
DX acceptance check: *a developer who has never seen LangBot can, from the docs
alone, push a message and observe a multi-part reply on their callback within
10 minutes.*
### Quickstart (curl)
```bash
BOT=https://your-langbot/bots/2f1c....
SECRET=supersecret
BODY='{"session_id":"ticket-10293","message":[{"type":"Plain","text":"hello"}]}'
TS=$(date +%s)
SIG="sha256=$(printf '%s.%s' "$TS" "$BODY" | openssl dgst -sha256 -hmac "$SECRET" -r | cut -d' ' -f1)"
curl -sS -X POST "$BOT" \
-H "Content-Type: application/json" \
-H "X-LB-Timestamp: $TS" \
-H "X-LB-Signature: $SIG" \
-d "$BODY"
```
---
## 13. Future work
- **Durable outbound queue** (persist + replay across restarts; exactly-once).
- **Per-caller API keys** with rotation and scopes (multi-tenant Space usage).
- **Sync convenience endpoint** (§9) once core is stable.
- **Server-Sent Events outbound option** for callers that *do* want a stream but
not a full duplex socket — single GET, server pushes chunks.
- **Dashboard "test console"** for `http_bot` (send a message, watch callbacks)
mirroring the existing WebSocket debug panel.
---
## 14. Rollout / task breakdown
| # | Task | Touches |
|---|---|---|
| 1 | `http_bot.yaml` manifest + icon | `platform/sources/` |
| 2 | `HttpBotAdapter` (inbound verify, event build, outbound queue) | `platform/sources/http_bot.py` |
| 3 | Signing helper module (shared) | `platform/sources/` or `utils/` |
| 4 | i18n strings (en/zh/ja) | adapter yaml + web locale |
| 5 | Integration docs `docs/platforms/http-bot.md` | `docs/` |
| 6 | OpenAPI fragment + reference clients | `docs/`, `examples/http-bot/` |
| 7 | Tests: signature verify, N→1 aggregation, 1→M ordering, retry | `tests/` |
| 8 | (opt) SSRF guard for callback_url | adapter |
No changes required to: the unified webhook router, the aggregator, the query
pool, or the pipeline. That is the design's main payoff.
---
## 15. Resolved decisions
1. **Callback URL trust****config-only.** The inbound message may not carry a
`callback_url`; replies always go to the bot-config URL. Closes the SSRF
vector where a leaked inbound secret could redirect replies.
2. **Session lifecycle****`POST /bots/<uuid>/reset`** (body `{session_id,
session_type?}`) drops the matching session from the session manager; the
next message starts a fresh conversation. Implemented via sub-path routing in
`handle_unified_webhook`.
3. **Group semantics** — for `session_type: group`, `session_id` is the group/
launcher id; `sender.id` (and optional `sender.group_name`) identify the
member. A Space ticket maps to one `session_id`.
4. **Backpressure** — bounded per-session outbound queue (maxlen 1000); on
overflow the oldest reply is dropped and a warning logged, so a persistently
down callback can never exhaust memory.
### Still open / deferred (see §13)
- Durable outbound queue (persist + replay across restarts).
- Per-caller API keys with rotation/scopes for multi-tenant Space usage.
- SSE outbound option and a dashboard test console.
-198
View File
@@ -1,198 +0,0 @@
{
"openapi": "3.0.3",
"info": {
"title": "LangBot HTTP Bot Adapter",
"version": "1.0.0",
"description": "Server-to-server HTTP integration for a LangBot pipeline. Inbound messages are POSTed to the unified webhook route; replies are delivered to a configured callback URL (one POST per reply part). All requests are HMAC-SHA256 signed. See docs/platforms/http-bot.md."
},
"paths": {
"/bots/{bot_uuid}": {
"post": {
"summary": "Push a message into the pipeline (fire-and-collect)",
"description": "Returns 202 immediately. Replies arrive asynchronously on the configured callback URL. Reuse the same session_id within the aggregation window to merge multiple messages into one turn (N->1).",
"parameters": [
{ "$ref": "#/components/parameters/BotUuid" },
{ "$ref": "#/components/parameters/Timestamp" },
{ "$ref": "#/components/parameters/Signature" },
{ "$ref": "#/components/parameters/Idempotency" }
],
"requestBody": {
"required": true,
"content": { "application/json": { "schema": { "$ref": "#/components/schemas/InboundMessage" } } }
},
"responses": {
"202": {
"description": "Accepted (queued for the pipeline)",
"content": { "application/json": { "schema": { "$ref": "#/components/schemas/AcceptedResponse" } } }
},
"400": { "$ref": "#/components/responses/Error" },
"401": { "$ref": "#/components/responses/Error" },
"409": { "$ref": "#/components/responses/Error" },
"413": { "$ref": "#/components/responses/Error" }
}
}
},
"/bots/{bot_uuid}/sync": {
"post": {
"summary": "Push a message and wait for the collapsed reply",
"description": "Blocking convenience mode. Waits for is_final and returns all reply parts collapsed into one array. Lossy (no sequence/streaming). One in-flight sync per session_id.",
"parameters": [
{ "$ref": "#/components/parameters/BotUuid" },
{ "$ref": "#/components/parameters/Timestamp" },
{ "$ref": "#/components/parameters/Signature" }
],
"requestBody": {
"required": true,
"content": { "application/json": { "schema": { "$ref": "#/components/schemas/InboundMessage" } } }
},
"responses": {
"200": {
"description": "The collapsed reply",
"content": { "application/json": { "schema": { "$ref": "#/components/schemas/SyncResponse" } } }
},
"400": { "$ref": "#/components/responses/Error" },
"401": { "$ref": "#/components/responses/Error" }
}
}
},
"/bots/{bot_uuid}/reset": {
"post": {
"summary": "Reset a session's conversation",
"parameters": [
{ "$ref": "#/components/parameters/BotUuid" },
{ "$ref": "#/components/parameters/Timestamp" },
{ "$ref": "#/components/parameters/Signature" }
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"type": "object",
"required": ["session_id"],
"properties": {
"session_id": { "type": "string" },
"session_type": { "type": "string", "enum": ["person", "group"] }
}
}
}
}
},
"responses": {
"200": { "description": "Reset done" },
"400": { "$ref": "#/components/responses/Error" },
"401": { "$ref": "#/components/responses/Error" }
}
}
}
},
"components": {
"parameters": {
"BotUuid": {
"name": "bot_uuid", "in": "path", "required": true,
"schema": { "type": "string", "format": "uuid" }
},
"Timestamp": {
"name": "X-LB-Timestamp", "in": "header", "required": true,
"description": "Unix seconds; rejected if more than +/-300s from server time.",
"schema": { "type": "string" }
},
"Signature": {
"name": "X-LB-Signature", "in": "header", "required": true,
"description": "sha256=<hex> of HMAC-SHA256(secret, \"{timestamp}.\" + raw_body).",
"schema": { "type": "string" }
},
"Idempotency": {
"name": "X-LB-Idempotency-Key", "in": "header", "required": false,
"description": "Dedup key; a repeat within the dedup window returns 409.",
"schema": { "type": "string" }
}
},
"schemas": {
"Segment": {
"type": "object",
"required": ["type"],
"properties": {
"type": { "type": "string", "enum": ["Plain", "Image", "Voice", "File", "At", "Quote"] },
"text": { "type": "string", "description": "For type=Plain." },
"url": { "type": "string", "description": "For media types." },
"base64": { "type": "string", "description": "For media types (data URI or raw base64)." }
}
},
"InboundMessage": {
"type": "object",
"required": ["session_id", "message"],
"properties": {
"session_id": { "type": "string", "description": "Caller-defined; maps 1:1 to a LangBot session." },
"session_type": { "type": "string", "enum": ["person", "group"], "default": "person" },
"sender": {
"type": "object",
"properties": {
"id": { "type": "string" },
"name": { "type": "string" },
"group_name": { "type": "string", "description": "For session_type=group." }
}
},
"message": { "type": "array", "items": { "$ref": "#/components/schemas/Segment" } }
}
},
"AcceptedResponse": {
"type": "object",
"properties": {
"code": { "type": "integer", "example": 0 },
"msg": { "type": "string", "example": "accepted" },
"data": {
"type": "object",
"properties": {
"session_id": { "type": "string" },
"accepted_message_id": { "type": "string", "example": "in_01H..." },
"aggregating": { "type": "boolean" }
}
}
}
},
"SyncResponse": {
"type": "object",
"properties": {
"code": { "type": "integer", "example": 0 },
"msg": { "type": "string", "example": "ok" },
"data": {
"type": "object",
"properties": {
"session_id": { "type": "string" },
"reply_to": { "type": "string" },
"message": { "type": "array", "items": { "$ref": "#/components/schemas/Segment" } }
}
}
}
},
"Callback": {
"type": "object",
"description": "Delivered by LangBot to your callback_url, one POST per reply part. Signed with the outbound secret.",
"properties": {
"session_id": { "type": "string" },
"reply_to": { "type": "string", "description": "The accepted_message_id this answers." },
"sequence": { "type": "integer", "description": "1-based ordinal within the turn." },
"is_final": { "type": "boolean", "description": "True on the last part of the turn." },
"stream": { "type": "boolean" },
"message": { "type": "array", "items": { "$ref": "#/components/schemas/Segment" } },
"timestamp": { "type": "string", "format": "date-time" }
}
},
"ErrorEnvelope": {
"type": "object",
"properties": {
"code": { "type": "integer", "example": 40101 },
"msg": { "type": "string", "example": "invalid signature: signature_mismatch" },
"data": { "nullable": true }
}
}
},
"responses": {
"Error": {
"description": "Error envelope",
"content": { "application/json": { "schema": { "$ref": "#/components/schemas/ErrorEnvelope" } } }
}
}
}
}
-256
View File
@@ -1,256 +0,0 @@
# HTTP Bot Adapter — Integration Guide
Integrate **any backend system** with a LangBot pipeline over plain HTTP. Push
messages in via a signed webhook; receive replies on a callback URL. No
long-lived connection, full support for message **aggregation** (many inbound
messages merged into one turn) and **multi-part replies** (one turn → many
outbound messages).
This is the right adapter for **server-to-server** integrations — ticketing
systems, CRMs, internal tools, custom web backends. (For an in-browser,
real-time chat widget, use the embeddable Web Page Bot instead.)
> **5-minute goal:** stand up a callback receiver, send a message, and watch a
> multi-part reply arrive — using the reference client in
> [`examples/http-bot/`](../../examples/http-bot/).
---
## 1. Mental model
```
Your backend ──(1) POST signed message──► LangBot /bots/<bot_uuid>
(pipeline runs: aggregate → think → reply)
Your callback ◄─(2) POST signed reply(s)── LangBot one POST per reply part
```
- **(1) Inbound** is *fire-and-collect*: LangBot answers `202 Accepted`
immediately and does **not** return the pipeline result on that response.
- **(2) Outbound** replies arrive later as separate signed POSTs to your
`callback_url`. A single turn may produce **several** callbacks (e.g. a tool
call narration followed by the final answer).
- Everything is keyed by a **`session_id` you choose** (e.g. a ticket number).
Each `session_id` maps to one isolated LangBot conversation.
---
## 2. Create the bot
1. In the LangBot dashboard, add a bot and choose the **HTTP Bot** platform.
2. Fill in the config:
| Field | Required | Notes |
|---|---|---|
| **Inbound Signing Secret** | yes | Your backend signs inbound requests with this. |
| **Outbound Callback URL** | yes | Where LangBot POSTs replies. **Config-only** — cannot be overridden per message (SSRF protection). |
| **Outbound Signing Secret** | no | LangBot signs callbacks with this; defaults to the inbound secret. |
| **Default Session Type** | no | `person` (default) or `group`. |
| **Require Inbound Signature** | no | Keep `true` in production. |
| **Callback Timeout / Max Retries** | no | Defaults: 15s, 3 retries. |
3. Bind the bot to a **pipeline** and **enable** it.
4. Copy the **Inbound Webhook URL** shown in the config — it looks like
`https://your-langbot/bots/<bot_uuid>`.
---
## 3. The signature scheme
Both directions use the same dependency-free HMAC-SHA256 scheme:
```
signing_string = "{timestamp}." + raw_body_bytes
signature = "sha256=" + hex(HMAC_SHA256(secret, signing_string))
```
Sent as headers:
| Header | Meaning |
|---|---|
| `X-LB-Timestamp` | Unix seconds. Rejected if more than **±300s** from server time. |
| `X-LB-Signature` | `sha256=<hex>` over `"{timestamp}." + body`. |
| `X-LB-Idempotency-Key` | *(optional, inbound)* dedup key; retries with the same key return `409`. |
Verify outbound callbacks the same way, using the **outbound** secret (or the
inbound secret if you left it blank).
A six-line reference implementation is in `examples/http-bot/client.py`
(`sign()` / `verify()`); a Node/TS version is in `client.ts`.
---
## 4. Send your first message (curl)
```bash
BOT="https://your-langbot/bots/<bot_uuid>"
SECRET="your-inbound-secret"
BODY='{"session_id":"ticket-10293","message":[{"type":"Plain","text":"Export keeps failing on the dashboard."}]}'
TS=$(date +%s)
SIG="sha256=$(printf '%s.%s' "$TS" "$BODY" | openssl dgst -sha256 -hmac "$SECRET" -r | cut -d' ' -f1)"
curl -sS -X POST "$BOT" \
-H "Content-Type: application/json" \
-H "X-LB-Timestamp: $TS" \
-H "X-LB-Signature: $SIG" \
-d "$BODY"
# -> 202 {"code":0,"msg":"accepted","data":{"session_id":"ticket-10293","accepted_message_id":"in_...","aggregating":true}}
```
The reply(s) will be POSTed to your configured callback URL shortly after.
---
## 5. Inbound request format
`POST /bots/{bot_uuid}`
```jsonc
{
"session_id": "ticket-10293", // REQUIRED. Your stable id. Maps 1:1 to a LangBot session.
"session_type": "person", // optional: "person" | "group"; default from config
"sender": { // optional metadata, surfaced to the pipeline/plugins
"id": "user-5567",
"name": "Alice"
},
"message": [ // REQUIRED. A LangBot MessageChain (array of segments).
{ "type": "Plain", "text": "Export keeps failing on the dashboard." },
{ "type": "Image", "url": "https://example.com/screenshot.png" }
]
}
```
**Message segments.** Text uses `{"type":"Plain","text":"..."}`. Images use
`{"type":"Image","url":"..."}` (or `base64`). Other supported types: `Voice`,
`File`, `At`, `Quote`.
> Note: the callback URL is **not** accepted in the body — it is taken only from
> bot config. This is deliberate (prevents an attacker who obtains the inbound
> secret from redirecting replies to an arbitrary host).
### Aggregation (N → 1)
If your pipeline has **message aggregation** enabled, send several messages with
the **same `session_id`** within the aggregation window and they are merged into
**one** pipeline turn. No special flag — just reuse the `session_id`.
---
## 6. Outbound callback format
LangBot POSTs each reply part to your `callback_url`:
```jsonc
{
"session_id": "ticket-10293", // echoes the inbound session
"reply_to": "in_01H...", // the accepted_message_id this answers
"sequence": 1, // 1-based ordinal within this turn
"is_final": false, // true on the last part of the turn
"stream": false, // true for streamed chunks
"message": [ { "type": "Plain", "text": "Looking into it…" } ],
"timestamp": "2026-06-22T09:00:01Z"
}
```
Your endpoint should return `2xx` quickly. Non-2xx / timeout → LangBot retries
with exponential backoff (up to `callback_max_retries`).
### Multi-part replies (1 → M)
One turn may emit multiple callbacks, delivered **in `sequence` order** for a
given session:
```
seq=1 is_final=false "Checking your export logs…"
seq=2 is_final=false "Found 2 failed exports."
seq=3 is_final=true "Fixed — please try again."
```
Stitch by `session_id` + `sequence`; the turn is complete when
`is_final: true` arrives.
---
## 7. Reset a session
Start a fresh conversation for a `session_id` (drops history):
```
POST /bots/{bot_uuid}/reset
{ "session_id": "ticket-10293", "session_type": "person" }
→ 200 { "code":0, "msg":"reset", "data": { "session_id":"ticket-10293", "removed": true } }
```
Signed exactly like an inbound message.
---
## 8. Synchronous convenience mode
If you don't need streaming/multi-part and just want one reply back on the same
HTTP call, POST to `/sync`. LangBot waits for the turn to finish and returns all
parts **collapsed** into one array:
```
POST /bots/{bot_uuid}/sync
{ "session_id": "ticket-10293", "message": [ { "type":"Plain", "text":"hi" } ] }
→ 200 { "code":0, "msg":"ok",
"data": { "session_id":"ticket-10293", "reply_to":"in_...",
"message": [ {"type":"Plain","text":"..."}, ... ] } }
```
This is **lossy** (you lose `sequence` / streaming boundaries) and blocks up to
`callback_timeout × 4` seconds. Prefer the callback model for anything
real-time or multi-part. Only one in-flight `/sync` per `session_id`.
---
## 9. Error envelope
```jsonc
{ "code": 40101, "msg": "invalid signature: signature_mismatch", "data": null }
```
| HTTP | code | meaning |
|---|---|---|
| 202 | 0 | accepted |
| 400 | 40001 | malformed body / missing `session_id` or `message` |
| 401 | 40101 | bad/expired signature |
| 409 | 40901 | duplicate idempotency key |
| 413 | 41301 | message too large (>1 MiB) |
| 500 | 50001 | internal error |
---
## 10. Try it end-to-end in 5 minutes
```bash
cd examples/http-bot
pip install flask requests
# Terminal 1 — your callback receiver (point the bot's callback_url here, e.g. via a tunnel):
python client.py serve --port 8900 --secret SHARED_SECRET
# Terminal 2 — push a message:
python client.py push \
--url https://your-langbot/bots/<bot_uuid> \
--secret SHARED_SECRET \
--session ticket-1 \
--text "hello"
```
Watch Terminal 1 print each reply part (`[part ]` / `[FINAL]`) with its
sequence number — that's 1→M working, signatures verified.
A machine-readable contract is in
[`docs/http-bot-openapi.json`](../http-bot-openapi.json).
---
## 11. Security checklist
- Keep **Require Inbound Signature** on in production.
- Use **HTTPS** callback URLs; the URL is config-only (no per-message override).
- Treat the secrets like passwords; rotate via the dashboard.
- The inbound route is unauthenticated at the framework level **by design**
security comes entirely from the HMAC signature, so never disable it on a
public deployment.
-196
View File
@@ -1,196 +0,0 @@
# MCP Resources PR #2215 Review
> 更新日期: 2026-06-29
> 分支: `mcp_resources`
> PR: langbot-app/LangBot#2215
> 主题: MCP Resources 在 LangBot 中的产品价值、AgentRunner 集成方式与后续架构方向
## 结论
PR #2215 对 LangBot 有明确价值:它补齐了 MCP 协议中 Resources 这一重要能力,让 MCP server 不再只暴露 tools,也可以暴露文档、代码片段、配置、日志、图片等上下文资源。管理端可以发现和预览资源,Agent 也可以通过当前实现按需列出和读取资源。
但当前 AgentRunner 层的接入方式更接近一个可用的第一阶段方案,而不是最终架构。现在 MCP Resources 被包装成两个 synthetic tools
- `langbot_mcp_list_resources`
- `langbot_mcp_read_resource`
这让模型可以通过 function calling 主动探索资源,落地成本低,也复用了已有 `ToolManager` / `LocalAgentRunner` 的工具调用链路。不过从 MCP 规范和主流实现来看,Resources 更适合作为一种一等上下文来源,而不是长期隐藏在工具列表里。
建议保留当前 synthetic tools 作为探索能力,同时把后续主线设计调整为:MCP Resources 是 pipeline / conversation / message 级别可选择、可固定、可审计的上下文输入。
## 当前实现判断
当前 AgentRunner 集成路径如下:
```text
Pipeline 绑定 MCP server
-> query.variables['_pipeline_bound_mcp_servers']
-> Preproc 为 local-agent 加载工具
-> ToolManager.get_all_tools()
-> MCPLoader 注入 synthetic resource tools
-> LocalAgentRunner 将工具 schema 传给模型
-> 模型发起 list/read tool call
-> ToolManager.execute_func_call()
-> MCPLoader 调 MCP session.list_resources/read_resource
-> tool result 回灌给模型
```
这个路径的优点是:
- 复用现有工具调用机制,改动范围小。
- Agent 可以按需探索资源,不需要每轮预先读取所有资源。
- 可以沿用 pipeline 绑定的 MCP server 范围,避免越权读取未绑定 server。
- 对已有 MCP tools 行为影响较小。
主要问题是:
- Resources 在语义上被降级成 tools,和 MCP 规范里的 resource primitive 不完全一致。
- 模型必须先理解并主动调用 `list/read`,资源不会自然成为上下文。
- pipeline 不能配置“默认携带某些资源”或“本轮附加某些资源”。
- UI 资源 tab 目前是管理端预览能力,和 Agent 上下文选择没有打通。
- 对 blob、图片、大文件、结构化资源的处理还比较粗糙。
- 缺少 resource templates、订阅更新、缓存、chunk、token budget、trace 与审计策略。
## 主流项目做法
### MCP 官方规范
MCP Resources 是 server 暴露上下文数据的协议能力。规范没有要求 resources 必须以 tool call 形式给模型使用,而是把如何选择、过滤、读取和纳入上下文交给 Host application。
这意味着比较正统的集成方式是:LangBot 作为 Host,在 pipeline、会话或消息层决定哪些 resources 进入模型上下文。
参考: https://modelcontextprotocol.io/specification/2025-06-18/server/resources
### VS Code Copilot
VS Code 把 MCP Resources 做成 chat context 的一部分。用户可以通过 `Add Context > MCP Resources` 或命令浏览 MCP resources,并把选中的资源附加到一次 chat request。
这是目前最值得 LangBot 参考的产品形态:资源不是模型工具,而是用户和 Host 可控的上下文附件。
参考: https://code.visualstudio.com/docs/agent-customization/mcp-servers
### Anthropic SDK
Anthropic 的 client-side MCP helpers 提供资源读取和转换能力,例如把 MCP resource 转为 Claude message content 或 file。也就是说,应用先读取 resource,再显式放进模型消息。
这同样是 application-owned context injection,而不是把 resource 伪装成模型工具。
参考: https://platform.claude.com/docs/en/agents-and-tools/mcp-connector
### LangChain MCP Adapters
LangChain 把 MCP Resources 更像 data loader / document input 来处理,可以把资源加载成 `Blob`,再进入 LangChain 的文档、检索或上下文处理链路。
这说明 Resources 很适合作为知识源、文档源或上下文源,而不只是即时工具调用。
参考: https://docs.langchain.com/oss/python/langchain/mcp
### OpenAI Agents SDK
OpenAI Agents SDK 主路径仍偏向 MCP tools,但底层 MCP server API 已经有 `list_resources``list_resource_templates``read_resource` 等能力。当前形态说明 resources 是 client 能力,但并未默认变成 agent-visible tools。
参考: https://openai.github.io/openai-agents-python/mcp/
### Cline
Cline 会拉取 MCP tools、resources、resourceTemplates、prompts,并通过类似 `access_mcp_resource` 的内置访问方式让模型读取资源。这个方向和 LangBot 当前 synthetic tools 比较接近。
这种模式适合让 Agent 自主探索,但更像 Host 自定义的模型访问协议,不应成为唯一集成路径。
参考: https://github.com/cline/cline/blob/main/src/services/mcp/McpHub.ts
## 建议架构方向
### 1. 保留探索型工具
保留当前两个 synthetic tools
- `langbot_mcp_list_resources`
- `langbot_mcp_read_resource`
它们适合处理“用户没有显式选择资源,但 Agent 判断需要探索 MCP server 上下文”的场景。后续可以优化工具描述、返回格式、资源大小限制和错误信息。
### 2. 增加一等 Resource Context
新增一个 Host 层资源上下文概念,例如:
```text
PipelineResourceBinding
ConversationResourceAttachment
MessageResourceAttachment
```
Preproc 或独立的 `ResourceContextProvider` 在模型调用前读取这些资源,按 MIME 类型、大小、token budget 转为模型可消费的上下文。
### 3. 打通 UI 与 Agent 上下文
当前 MCP 详情页的 Resources tab 可以继续作为资源发现和预览入口。建议增加操作:
- 添加到本轮上下文
- 固定到当前 pipeline
- 固定到当前 bot / conversation
- 查看资源读取历史和错误
这样 UI 资源管理能力才能真正影响 Agent 行为。
### 4. 支持 resource templates
MCP resource templates 允许 server 暴露参数化资源,例如:
```text
repo://{owner}/{repo}/file/{path}
log://{service}/{date}
```
LangBot 后续应支持模板发现、参数填写、实例化和绑定。否则只能使用静态 resources,覆盖面会受限。
### 5. 增加资源处理策略
建议补齐:
- 文本资源 token budget 与截断策略。
- 大文件 chunk 与摘要策略。
- 图片/blob 的模型能力判断与 fallback。
- MIME 类型白名单与安全限制。
- 缓存与过期策略。
- `resources/listChanged` 或订阅更新。
- resource read trace,便于审计 Agent 读取了什么上下文。
## 推荐落地顺序
### Phase 1: 完成当前 PR 可用性
- 保留 synthetic tools。
- 明确文档说明当前 Agent 集成是 tool-mediated。
- 完善资源工具描述,降低模型误用概率。
- 给 read/list 增加大小限制和更清晰的 MIME 处理。
- 前端 Resources tab 与 Tools tab 分离,保持管理端清晰。
### Phase 2: 做 Host-owned context attachments
- 在 pipeline 或 conversation 层新增 resource attachment 配置。
- Preproc 读取已绑定 resources,注入模型上下文。
- UI 支持“添加到上下文 / 固定到 pipeline”。
- 记录每轮实际注入的 resource URI 和 token 消耗。
### Phase 3: 做完整 MCP Resources 能力
- 支持 resource templates。
- 支持资源订阅更新。
- 支持 chunk、summary、RAG 化接入。
- 为 DifyAgentRunner、LocalAgentRunner 等不同 runner 定义统一资源上下文接口。
## 最终建议
PR #2215 可以作为 MCP Resources 的第一阶段实现继续推进。它让 LangBot 快速拥有“资源发现、预览、按需读取”的闭环,也给 Agent 探索资源提供了可运行路径。
但在正式设计上,不建议把 “Resources == Tools” 固化为长期抽象。LangBot 更应该把 MCP Resources 定位为上下文来源,与 tools、prompts、knowledge base 并列:
```text
Tools -> Agent 可以执行的动作
Resources -> Host/用户/Agent 可以选择的上下文数据
Prompts -> 可复用的任务模板
Knowledge -> 可检索、可索引的长期知识
```
这样既尊重 MCP 协议语义,也能让 LangBot 在 Agent 工作流、企业知识接入和多 MCP server 管理上走得更稳。
-75
View File
@@ -1,75 +0,0 @@
# HTTP Bot Adapter — Reference Clients
> English | [中文](./README.zh.md)
Minimal, dependency-light clients for the LangBot **HTTP Bot** platform adapter.
They show the whole loop: signing a request, pushing a message, and receiving
multi-part replies on a callback endpoint.
Full guide: [docs.langbot.app — HTTP Bot](https://docs.langbot.app/en/usage/platforms/http-bot).
Machine-readable contract: [`docs/http-bot-openapi.json`](../../docs/http-bot-openapi.json).
## Files
| File | What it is |
|---|---|
| `playground.py` | **Interactive browser debug console** — a single-file web app you open in a browser to chat with a running `http_bot` bot and watch signing / 202 / callbacks live. Zero extra deps. |
| `client.py` | Python client + Flask callback receiver (`pip install flask requests`). |
| `client.ts` | TypeScript/Node 18+ client + callback receiver, **zero deps** (`npx tsx client.ts`). |
All three implement the identical HMAC-SHA256 scheme
(`sha256=hex(HMAC(secret, "{timestamp}." + body))`) — verified byte-for-byte
against the adapter.
## Interactive playground (recommended first run)
A self-contained web console: type a message in your browser, it is signed and
POSTed to a **running** `http_bot` bot, and the bot's replies stream back into
the page — with a debug panel showing the signature, the `202` ack, and each
callback's `sequence` / signature-verification.
```bash
# From the LangBot repo root, with the backend already running:
PUBLIC_IP=<your-host-ip> ./.venv/bin/python examples/http-bot/playground.py
# then open http://<your-host-ip>:8920/
```
On startup it reads the LangBot API key + the `http_bot` bot from
`data/langbot.db`, and configures that bot (inbound/outbound secret +
`callback_url`) to point back at itself via the LangBot API — the bot reloads
live, no restart needed. Requirements: an enabled `http_bot` bot bound to a
working pipeline, and port `8920` reachable from your browser.
Env knobs: `PUBLIC_IP` (default `127.0.0.1`), `PLAYGROUND_PORT` (default `8920`).
## Headless clients
```bash
# Python — Terminal 1: callback receiver (your callback_url target)
python client.py serve --port 8900 --secret SHARED_SECRET
# Python — Terminal 2: push a message
python client.py push --url https://your-langbot/bots/<BOT_UUID> \
--secret SHARED_SECRET --session ticket-1 --text "hello"
# blocking sync mode
python client.py sync --url https://your-langbot/bots/<BOT_UUID> \
--secret SHARED_SECRET --session ticket-1 --text "hello"
# reset a session
python client.py reset --url https://your-langbot/bots/<BOT_UUID> \
--secret SHARED_SECRET --session ticket-1
```
```bash
# TypeScript (Node 18+)
npx tsx client.ts serve 8900 SHARED_SECRET
npx tsx client.ts push https://your-langbot/bots/<BOT_UUID> SHARED_SECRET ticket-1 "hello"
```
When the bot replies, the receiver prints each part with its `sequence` and an
`[FINAL]` marker on the last one — that's the 1→M multi-reply model in action.
> The bot's `callback_url` must be reachable from LangBot. For local testing,
> expose your receiver with a tunnel (cloudflared / ngrok) and set that URL in
> the bot config.
-71
View File
@@ -1,71 +0,0 @@
# HTTP Bot 适配器 —— 参考客户端
> [English](./README.md) | 中文
面向 LangBot **HTTP Bot** 平台适配器的极简、低依赖客户端示例。
它们完整展示了整条链路:对请求签名、推送一条消息、在回调端点接收
1→M 的多段回复。
完整指南:[docs.langbot.app —— HTTP Bot](https://docs.langbot.app/zh/usage/platforms/http-bot)。
机器可读的接口契约:[`docs/http-bot-openapi.json`](../../docs/http-bot-openapi.json)。
## 文件清单
| 文件 | 是什么 |
|---|---|
| `playground.py` | **浏览器交互式调试台** —— 单文件 Web 应用,在浏览器里和一个运行中的 `http_bot` bot 对话,实时观察签名 / 202 / 回调。零额外依赖。 |
| `client.py` | Python 客户端 + Flask 回调接收端(`pip install flask requests`)。 |
| `client.ts` | TypeScript/Node 18+ 客户端 + 回调接收端,**零依赖**(`npx tsx client.ts`)。 |
三者实现完全一致的 HMAC-SHA256 签名方案
(`sha256=hex(HMAC(secret, "{timestamp}." + body))`)—— 已与适配器逐字节比对验证。
## 交互式 playground(推荐先跑这个)
一个自包含的 Web 控制台:在浏览器里输入消息,它会被签名并 POST 给一个
**运行中**的 `http_bot` bot,bot 的回复会流式回到页面上 —— 调试面板会显示
签名、`202` 确认,以及每条回调的 `sequence` / 签名验证结果。
```bash
# 在 LangBot 仓库根目录、后端已启动的前提下:
PUBLIC_IP=<你的主机IP> ./.venv/bin/python examples/http-bot/playground.py
# 然后打开 http://<你的主机IP>:8920/
```
启动时它会从 `data/langbot.db` 读取 LangBot API key 和 `http_bot` bot,
并通过 LangBot API 把该 bot 配好(入站/出站密钥 + `callback_url`)指回自己 ——
bot 会热加载,无需重启。前提:有一个已启用、绑定了可用 pipeline 的
`http_bot` bot,且端口 `8920` 能从你的浏览器访问到。
可调环境变量:`PUBLIC_IP`(默认 `127.0.0.1`)、`PLAYGROUND_PORT`(默认 `8920`)。
## 无头客户端
```bash
# Python —— 终端 1:回调接收端(你的 callback_url 指向它)
python client.py serve --port 8900 --secret SHARED_SECRET
# Python —— 终端 2:推送一条消息
python client.py push --url https://your-langbot/bots/<BOT_UUID> \
--secret SHARED_SECRET --session ticket-1 --text "hello"
# 阻塞式同步模式
python client.py sync --url https://your-langbot/bots/<BOT_UUID> \
--secret SHARED_SECRET --session ticket-1 --text "hello"
# 重置一个会话
python client.py reset --url https://your-langbot/bots/<BOT_UUID> \
--secret SHARED_SECRET --session ticket-1
```
```bash
# TypeScript(Node 18+)
npx tsx client.ts serve 8900 SHARED_SECRET
npx tsx client.ts push https://your-langbot/bots/<BOT_UUID> SHARED_SECRET ticket-1 "hello"
```
当 bot 回复时,接收端会逐条打印,带上各自的 `sequence`,并在最后一条标记
`[FINAL]` —— 这就是 1→M 多段回复模型的实际效果。
> bot 的 `callback_url` 必须能从 LangBot 访问到。本地测试时,可用隧道
> (cloudflared / ngrok)把你的接收端暴露出去,并把那个 URL 填进 bot 配置。
-167
View File
@@ -1,167 +0,0 @@
#!/usr/bin/env python3
"""LangBot HTTP Bot adapter — reference client (Python).
Two things in one file:
1. ``push()`` / ``push_sync()`` send a message into a LangBot ``http_bot`` bot.
2. A tiny Flask callback receiver that verifies signatures and prints replies,
so you can watch N->1 aggregation and 1->M multi-reply working live.
Usage
-----
pip install flask requests
# Terminal 1 — start the callback receiver (this is your callback_url):
python client.py serve --port 8900 --secret SHARED_SECRET
# Terminal 2 — push a message (async; reply lands on the receiver):
python client.py push \
--url https://your-langbot/bots/<BOT_UUID> \
--secret SHARED_SECRET \
--session ticket-10293 \
--text "Export keeps failing on the dashboard."
# Or push and block for the collapsed reply (sync convenience mode):
python client.py sync --url https://your-langbot/bots/<BOT_UUID> \
--secret SHARED_SECRET --session ticket-10293 --text "hi"
The signing scheme is HMAC-SHA256 over ``"{timestamp}." + raw_body``; see
``sign()`` below it is intentionally tiny and easy to port.
"""
from __future__ import annotations
import argparse
import hashlib
import hmac
import json
import sys
import time
import uuid
HEADER_TIMESTAMP = 'X-LB-Timestamp'
HEADER_SIGNATURE = 'X-LB-Signature'
HEADER_IDEMPOTENCY = 'X-LB-Idempotency-Key'
REPLAY_WINDOW = 300
def sign(secret: str, body: bytes, timestamp: int | None = None) -> tuple[str, str]:
"""Return (timestamp, signature) for *body*."""
ts = str(timestamp if timestamp is not None else int(time.time()))
mac = hmac.new(secret.encode(), f'{ts}.'.encode() + body, hashlib.sha256)
return ts, 'sha256=' + mac.hexdigest()
def verify(secret: str, body: bytes, timestamp: str | None, signature: str | None) -> bool:
"""Verify an inbound signature (used by the callback receiver)."""
if not timestamp or not signature:
return False
try:
if abs(int(time.time()) - int(float(timestamp))) > REPLAY_WINDOW:
return False
except ValueError:
return False
_, expected = sign(secret, body, int(float(timestamp)))
return hmac.compare_digest(expected, signature)
def _post(url: str, secret: str, payload: dict, idempotency: bool = True):
import requests
body = json.dumps(payload, ensure_ascii=False).encode()
ts, sig = sign(secret, body)
headers = {
'Content-Type': 'application/json',
HEADER_TIMESTAMP: ts,
HEADER_SIGNATURE: sig,
}
if idempotency:
headers[HEADER_IDEMPOTENCY] = uuid.uuid4().hex
resp = requests.post(url, data=body, headers=headers, timeout=30)
print(f'-> {resp.status_code} {resp.text}')
return resp
def push(url: str, secret: str, session: str, text: str, session_type: str = 'person'):
"""Fire-and-collect: returns 202 immediately; reply arrives on your callback."""
payload = {
'session_id': session,
'session_type': session_type,
'message': [{'type': 'Plain', 'text': text}],
}
return _post(url.rstrip('/'), secret, payload)
def push_sync(url: str, secret: str, session: str, text: str, session_type: str = 'person'):
"""Blocking convenience: POST to /sync and get the collapsed reply back."""
payload = {
'session_id': session,
'session_type': session_type,
'message': [{'type': 'Plain', 'text': text}],
}
resp = _post(url.rstrip('/') + '/sync', secret, payload, idempotency=False)
return resp
def reset(url: str, secret: str, session: str, session_type: str = 'person'):
"""Reset a session's conversation (next message starts fresh)."""
payload = {'session_id': session, 'session_type': session_type}
return _post(url.rstrip('/') + '/reset', secret, payload, idempotency=False)
def serve(port: int, secret: str):
"""Run a callback receiver that verifies signatures and prints replies."""
from flask import Flask, request
app = Flask(__name__)
@app.route('/', methods=['POST'])
def recv():
raw = request.get_data()
ok = verify(secret, raw, request.headers.get(HEADER_TIMESTAMP), request.headers.get(HEADER_SIGNATURE))
if not ok:
print('!! signature verification FAILED — rejecting')
return {'error': 'bad signature'}, 401
data = json.loads(raw)
text_parts = [c.get('text', '') for c in data.get('message', []) if c.get('type') == 'Plain']
marker = 'FINAL' if data.get('is_final') else 'part '
print(
f'[{marker}] session={data["session_id"]} seq={data["sequence"]} '
f'reply_to={data.get("reply_to")}: {" ".join(text_parts)}'
)
return {'ok': True}
print(f'callback receiver listening on http://0.0.0.0:{port}/ (Ctrl-C to stop)')
app.run(host='0.0.0.0', port=port)
def main(argv=None):
p = argparse.ArgumentParser(description='LangBot HTTP Bot reference client')
sub = p.add_subparsers(dest='cmd', required=True)
sp = sub.add_parser('serve', help='run the callback receiver')
sp.add_argument('--port', type=int, default=8900)
sp.add_argument('--secret', required=True)
for name in ('push', 'sync', 'reset'):
c = sub.add_parser(name)
c.add_argument('--url', required=True, help='https://host/bots/<BOT_UUID>')
c.add_argument('--secret', required=True)
c.add_argument('--session', required=True)
c.add_argument('--session-type', default='person', choices=['person', 'group'])
if name != 'reset':
c.add_argument('--text', required=True)
args = p.parse_args(argv)
if args.cmd == 'serve':
serve(args.port, args.secret)
elif args.cmd == 'push':
push(args.url, args.secret, args.session, args.text, args.session_type)
elif args.cmd == 'sync':
push_sync(args.url, args.secret, args.session, args.text, args.session_type)
elif args.cmd == 'reset':
reset(args.url, args.secret, args.session, args.session_type)
if __name__ == '__main__':
sys.exit(main())
-123
View File
@@ -1,123 +0,0 @@
/**
* LangBot HTTP Bot adapter reference client (TypeScript / Node 18+).
*
* Zero runtime dependencies (uses global `fetch`, `crypto`, and `http`).
*
* - `push()` : fire-and-collect; reply lands on your callback URL.
* - `pushSync()` : POST /sync and await the collapsed reply.
* - `reset()` : reset a session's conversation.
* - `startReceiver()` : a callback server that verifies signatures and logs
* replies, so you can watch N->1 and 1->M live.
*
* Run the demos:
* npx tsx client.ts serve 8900 SHARED_SECRET
* npx tsx client.ts push https://host/bots/<UUID> SHARED_SECRET ticket-1 "hello"
* npx tsx client.ts sync https://host/bots/<UUID> SHARED_SECRET ticket-1 "hello"
* npx tsx client.ts reset https://host/bots/<UUID> SHARED_SECRET ticket-1
*/
import { createHmac, randomUUID, timingSafeEqual } from 'node:crypto';
import { createServer } from 'node:http';
const HEADER_TIMESTAMP = 'X-LB-Timestamp';
const HEADER_SIGNATURE = 'X-LB-Signature';
const HEADER_IDEMPOTENCY = 'X-LB-Idempotency-Key';
const REPLAY_WINDOW = 300;
/** Compute the `sha256=<hex>` signature over `"{ts}." + body`. */
export function sign(secret: string, body: Buffer | string, timestamp?: number): [string, string] {
const ts = String(timestamp ?? Math.floor(Date.now() / 1000));
const buf = typeof body === 'string' ? Buffer.from(body) : body;
const mac = createHmac('sha256', secret).update(Buffer.concat([Buffer.from(`${ts}.`), buf])).digest('hex');
return [ts, `sha256=${mac}`];
}
/** Verify an inbound signature (used by the callback receiver). */
export function verify(secret: string, body: Buffer, timestamp?: string, signature?: string): boolean {
if (!timestamp || !signature) return false;
if (Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp)) > REPLAY_WINDOW) return false;
const [, expected] = sign(secret, body, Number(timestamp));
const a = Buffer.from(expected);
const b = Buffer.from(signature);
return a.length === b.length && timingSafeEqual(a, b);
}
interface Segment { type: string; text?: string; url?: string; [k: string]: unknown }
async function post(url: string, secret: string, payload: object, idempotency = true) {
const body = Buffer.from(JSON.stringify(payload));
const [ts, sig] = sign(secret, body);
const headers: Record<string, string> = {
'Content-Type': 'application/json',
[HEADER_TIMESTAMP]: ts,
[HEADER_SIGNATURE]: sig,
};
if (idempotency) headers[HEADER_IDEMPOTENCY] = randomUUID();
const resp = await fetch(url, { method: 'POST', headers, body });
const text = await resp.text();
console.log(`-> ${resp.status} ${text}`);
return { status: resp.status, text };
}
/** Fire-and-collect: 202 now, reply later on your callback URL. */
export function push(url: string, secret: string, session: string, text: string, sessionType = 'person') {
return post(url.replace(/\/$/, ''), secret, {
session_id: session,
session_type: sessionType,
message: [{ type: 'Plain', text }] as Segment[],
});
}
/** Blocking convenience: POST /sync, get the collapsed reply. */
export function pushSync(url: string, secret: string, session: string, text: string, sessionType = 'person') {
return post(`${url.replace(/\/$/, '')}/sync`, secret, {
session_id: session,
session_type: sessionType,
message: [{ type: 'Plain', text }] as Segment[],
}, false);
}
/** Reset a session's conversation. */
export function reset(url: string, secret: string, session: string, sessionType = 'person') {
return post(`${url.replace(/\/$/, '')}/reset`, secret, { session_id: session, session_type: sessionType }, false);
}
/** Run a callback receiver that verifies signatures and prints replies. */
export function startReceiver(port: number, secret: string) {
const server = createServer((req, res) => {
if (req.method !== 'POST') { res.writeHead(405).end(); return; }
const chunks: Buffer[] = [];
req.on('data', (c) => chunks.push(c));
req.on('end', () => {
const raw = Buffer.concat(chunks);
const ok = verify(secret, raw, req.headers[HEADER_TIMESTAMP.toLowerCase()] as string,
req.headers[HEADER_SIGNATURE.toLowerCase()] as string);
if (!ok) {
console.log('!! signature verification FAILED — rejecting');
res.writeHead(401, { 'Content-Type': 'application/json' }).end(JSON.stringify({ error: 'bad signature' }));
return;
}
const data = JSON.parse(raw.toString());
const parts = (data.message as Segment[]).filter((c) => c.type === 'Plain').map((c) => c.text).join(' ');
const marker = data.is_final ? 'FINAL' : 'part ';
console.log(`[${marker}] session=${data.session_id} seq=${data.sequence} reply_to=${data.reply_to}: ${parts}`);
res.writeHead(200, { 'Content-Type': 'application/json' }).end(JSON.stringify({ ok: true }));
});
});
server.listen(port, () => console.log(`callback receiver listening on http://0.0.0.0:${port}/ (Ctrl-C to stop)`));
}
// --- CLI ---
const [cmd, ...rest] = process.argv.slice(2);
if (cmd === 'serve') {
startReceiver(Number(rest[0] ?? 8900), rest[1] ?? 'SHARED_SECRET');
} else if (cmd === 'push') {
push(rest[0], rest[1], rest[2], rest[3]);
} else if (cmd === 'sync') {
pushSync(rest[0], rest[1], rest[2], rest[3]);
} else if (cmd === 'reset') {
reset(rest[0], rest[1], rest[2]);
} else if (cmd) {
console.error(`unknown command: ${cmd}`);
process.exit(1);
}
-349
View File
@@ -1,349 +0,0 @@
#!/usr/bin/env python3
"""LangBot HTTP Bot — interactive playground (public, browser-based).
This is a REAL end-to-end demo against the RUNNING LangBot instance on this
host. It is NOT a mock and NOT an in-process import: every message you type in
the browser is signed and POSTed to the live `http_bot` bot at
http://127.0.0.1:5300/bots/<uuid>, and the bot's replies come back to this
server's /callback endpoint over real HTTP, then stream to your browser via SSE.
What it does on startup:
1. Reads the LangBot API key + the http_bot bot from data/langbot.db.
2. Configures the bot via the LangBot API (PUT /api/v1/platform/bots/<uuid>):
sets inbound_secret + outbound_secret + callback_url to point back here.
(LangBot reloads the bot live no server restart needed.)
3. Serves a chat page on 0.0.0.0:<PORT> so you can open it from the internet.
Run: ./.venv/bin/python examples/http-bot/playground.py
Then open: http://<this-host-public-ip>:<PORT>/
"""
from __future__ import annotations
import asyncio
import json
import os
import sqlite3
import sys
REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
sys.path.insert(0, os.path.join(REPO, 'src'))
from aiohttp import web # noqa: E402
import aiohttp # noqa: E402
from langbot.pkg.platform.sources import http_bot_signing as sg # noqa: E402
# ---- config -----------------------------------------------------------------
LANGBOT_BASE = 'http://127.0.0.1:5300'
DB_PATH = os.path.join(REPO, 'data', 'langbot.db')
PUBLIC_IP = os.environ.get('PUBLIC_IP', '127.0.0.1')
PORT = int(os.environ.get('PLAYGROUND_PORT', '8920'))
SECRET = 'playground-shared-secret'
# SSE subscribers: list of asyncio.Queue
subscribers: list[asyncio.Queue] = []
def db_lookup() -> tuple[str, str]:
"""Return (api_key, http_bot_uuid) from the LangBot DB."""
db = sqlite3.connect(DB_PATH)
db.row_factory = sqlite3.Row
api_key = db.execute('SELECT key FROM api_keys LIMIT 1').fetchone()['key']
bot = db.execute("SELECT uuid FROM bots WHERE adapter='http_bot' LIMIT 1").fetchone()
if not bot:
raise SystemExit('No http_bot bot found. Create one in the WebUI first.')
return api_key, bot['uuid']
async def configure_bot(api_key: str, bot_uuid: str, callback_url: str):
"""Point the live bot at this playground via the LangBot API.
update_bot() runs a raw SQL UPDATE with whatever keys we send, so we send a
MINIMAL payload: only adapter_config (built from scratch, not read back
the GET masks secrets). LangBot reloads + reruns the bot live.
"""
cfg = {
'inbound_secret': SECRET,
'outbound_secret': SECRET,
'callback_url': callback_url,
'signature_required': True,
'default_session_type': 'person',
'callback_timeout': 15,
'callback_max_retries': 3,
}
async with aiohttp.ClientSession() as s:
async with s.put(
f'{LANGBOT_BASE}/api/v1/platform/bots/{bot_uuid}',
headers={'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'},
json={'adapter_config': cfg},
) as r:
txt = await r.text()
print(f'[configure] PUT adapter_config -> {r.status} {txt[:200]}')
return r.status < 400
async def broadcast(event: dict):
for q in list(subscribers):
try:
q.put_nowait(event)
except Exception:
pass
# ---- HTTP handlers ----------------------------------------------------------
async def index(request: web.Request):
return web.Response(text=PAGE, content_type='text/html')
async def send(request: web.Request):
"""Browser -> here -> signed POST -> live LangBot bot."""
body_in = await request.json()
session_id = body_in.get('session_id') or 'playground-1'
text = body_in.get('text', '')
bot_uuid = request.app['bot_uuid']
payload = {
'session_id': session_id,
'sender': {'id': 'browser-user', 'name': 'You'},
'message': [{'type': 'Plain', 'text': text}],
}
raw = json.dumps(payload, ensure_ascii=False).encode()
ts, sig = sg.sign(SECRET, raw)
url = f'{LANGBOT_BASE}/bots/{bot_uuid}'
# echo what we send to the browser timeline
await broadcast(
{'dir': 'out', 'kind': 'request', 'session_id': session_id, 'text': text, 'url': url, 'sig': sig[:24] + ''}
)
async with aiohttp.ClientSession() as s:
async with s.post(
url,
data=raw,
headers={
'Content-Type': 'application/json',
sg.HEADER_TIMESTAMP: ts,
sg.HEADER_SIGNATURE: sig,
},
) as r:
status = r.status
try:
jr = await r.json()
except Exception:
jr = {'raw': await r.text()}
await broadcast({'dir': 'in', 'kind': 'ack', 'status': status, 'data': jr})
return web.json_response({'status': status, 'data': jr})
async def callback(request: web.Request):
"""Live LangBot bot -> here. Verify signature, stream to browser."""
raw = await request.read()
ok, why = sg.verify(SECRET, raw, request.headers.get(sg.HEADER_TIMESTAMP), request.headers.get(sg.HEADER_SIGNATURE))
data = json.loads(raw)
text = ' '.join(c.get('text', '') for c in data.get('message', []) if c.get('type') == 'Plain')
await broadcast(
{
'dir': 'in',
'kind': 'reply',
'session_id': data.get('session_id'),
'sequence': data.get('sequence'),
'is_final': data.get('is_final'),
'sig_ok': ok,
'sig_why': why,
'text': text,
}
)
return web.json_response({'ok': True})
async def events(request: web.Request):
"""SSE stream to the browser."""
resp = web.StreamResponse(
headers={
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
'Access-Control-Allow-Origin': '*',
}
)
await resp.prepare(request)
q: asyncio.Queue = asyncio.Queue()
subscribers.append(q)
try:
await resp.write(b': connected\n\n')
while True:
try:
ev = await asyncio.wait_for(q.get(), timeout=15)
await resp.write(f'data: {json.dumps(ev, ensure_ascii=False)}\n\n'.encode())
except asyncio.TimeoutError:
await resp.write(b': ping\n\n')
except (asyncio.CancelledError, ConnectionResetError):
pass
finally:
if q in subscribers:
subscribers.remove(q)
return resp
PAGE = r"""<!doctype html>
<html lang="zh"><head><meta charset="utf-8"/>
<meta name="viewport" content="width=device-width,initial-scale=1"/>
<title>LangBot HTTP Bot · 调试台</title>
<style>
:root{
--bg:#f7f8fa; --panel:#ffffff; --line:#e8eaed; --ink:#1f2329; --mut:#8a909a;
--brand:#2563eb; --brand-soft:#eef3ff; --ok:#16a34a; --bad:#dc2626; --code:#f3f4f6;
}
*{box-sizing:border-box}
html,body{height:100%}
body{margin:0;background:var(--bg);color:var(--ink);
font:14px/1.6 -apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,"PingFang SC","Microsoft YaHei",sans-serif}
.top{height:52px;background:var(--panel);border-bottom:1px solid var(--line);
display:flex;align-items:center;gap:10px;padding:0 18px}
.logo{width:26px;height:26px;border-radius:7px;background:var(--brand);display:grid;place-items:center;color:#fff;font-weight:700;font-size:14px}
.top b{font-size:15px} .top .ver{font-size:12px;color:var(--mut)}
.dot{width:8px;height:8px;border-radius:50%;background:#cbd2dc;display:inline-block;margin-right:5px;vertical-align:middle}
.dot.on{background:var(--ok)} .dot.off{background:var(--bad)}
.conn{margin-left:auto;font-size:12px;color:var(--mut)}
.wrap{max-width:1080px;margin:0 auto;padding:18px;display:grid;grid-template-columns:1fr 360px;gap:16px}
@media(max-width:880px){.wrap{grid-template-columns:1fr}}
.card{background:var(--panel);border:1px solid var(--line);border-radius:12px;display:flex;flex-direction:column;min-height:0}
.card h3{margin:0;padding:12px 16px;font-size:13px;font-weight:600;color:#4b5563;border-bottom:1px solid var(--line);display:flex;align-items:center;gap:8px}
.chat{height:62vh}
.msgs{flex:1;overflow:auto;padding:16px;display:flex;flex-direction:column;gap:12px}
.row{display:flex;flex-direction:column;gap:4px;max-width:82%}
.row.me{align-self:flex-end;align-items:flex-end}
.row.bot{align-self:flex-start}
.bub{padding:9px 13px;border-radius:12px;white-space:pre-wrap;word-break:break-word}
.me .bub{background:var(--brand);color:#fff;border-bottom-right-radius:3px}
.bot .bub{background:#f1f3f6;color:var(--ink);border-bottom-left-radius:3px}
.meta{font-size:11px;color:var(--mut)}
.meta .ok{color:var(--ok)} .meta .bad{color:var(--bad)}
.sys{align-self:center;font-size:12px;color:var(--mut);background:#f1f3f6;border-radius:8px;padding:4px 12px}
.bar{display:flex;gap:8px;padding:12px;border-top:1px solid var(--line)}
.bar input{flex:1;border:1px solid var(--line);border-radius:9px;padding:10px 12px;font-size:14px;outline:none}
.bar input:focus{border-color:var(--brand);box-shadow:0 0 0 3px var(--brand-soft)}
.bar button{background:var(--brand);color:#fff;border:0;border-radius:9px;padding:0 18px;font-size:14px;font-weight:500;cursor:pointer}
.bar button:disabled{opacity:.5;cursor:default}
.side{height:62vh}
.kv{padding:12px 16px;border-bottom:1px solid var(--line);font-size:12px}
.kv .k{color:var(--mut)} .kv .v{color:var(--ink);word-break:break-all}
.kv code{background:var(--code);border-radius:5px;padding:1px 5px;font-size:11px}
.sessrow{display:flex;align-items:center;gap:8px;padding:10px 16px;border-bottom:1px solid var(--line);font-size:12px}
.sessrow input{flex:1;border:1px solid var(--line);border-radius:7px;padding:5px 8px;font-size:12px}
.sessrow button{border:1px solid var(--line);background:#fff;border-radius:7px;padding:5px 9px;font-size:12px;cursor:pointer;color:#4b5563}
.trace{flex:1;overflow:auto;padding:10px 12px;font:11px/1.55 ui-monospace,SFMono-Regular,Menlo,monospace}
.ev{padding:6px 8px;border-radius:7px;margin-bottom:6px;border:1px solid var(--line)}
.ev .t{font-weight:600;font-size:10px;letter-spacing:.3px;text-transform:uppercase}
.ev.out{background:#f5f8ff;border-color:#dbe6ff}.ev.out .t{color:var(--brand)}
.ev.ack{background:#f4f6f8}.ev.ack .t{color:#6b7280}
.ev.reply{background:#f1faf3;border-color:#cdeed6}.ev.reply .t{color:var(--ok)}
.ev pre{margin:3px 0 0;white-space:pre-wrap;word-break:break-all;color:#374151}
</style></head>
<body>
<div class="top">
<div class="logo">L</div>
<b>HTTP Bot 调试台</b><span class="ver">examples/http-bot</span>
<span class="conn"><span class="dot off" id="cdot"></span><span id="conn">连接中</span></span>
</div>
<div class="wrap">
<!-- chat -->
<div class="card chat">
<h3>对话 · 真实发往运行中的 http_bot</h3>
<div class="msgs" id="msgs"></div>
<div class="bar">
<input id="msg" placeholder="输入消息,回车发送…" autofocus/>
<button id="send">发送</button>
</div>
</div>
<!-- debug -->
<div class="card side">
<h3>调试信息</h3>
<div class="kv"><span class="k">入站地址</span><br><span class="v"><code id="endpoint">/bots/&lt;uuid&gt;</code></span></div>
<div class="kv"><span class="k">签名</span> <span class="v">HMAC-SHA256 · <code>X-LB-Signature</code></span></div>
<div class="sessrow">
<span class="k">会话</span>
<input id="sid" value="playground-1"/>
<button id="reset">新会话</button>
</div>
<div class="trace" id="trace"></div>
</div>
</div>
<script>
const $=s=>document.querySelector(s);
const msgs=$('#msgs'),trace=$('#trace'),inp=$('#msg'),btn=$('#send'),
conn=$('#conn'),cdot=$('#cdot'),sidIn=$('#sid');
function el(c){const d=document.createElement('div');d.className=c;return d}
function atBottom(n){n.scrollTop=n.scrollHeight}
function bubble(side,text,metaHtml){
const r=el('row '+side),b=el('bub');b.textContent=text;r.appendChild(b);
if(metaHtml){const m=el('meta');m.innerHTML=metaHtml;r.appendChild(m)}
msgs.appendChild(r);atBottom(msgs)}
function sys(t){const d=el('sys');d.textContent=t;msgs.appendChild(d);atBottom(msgs)}
function logEv(kind,title,obj){
const e=el('ev '+kind),t=el('t');t.textContent=title;e.appendChild(t);
if(obj!==undefined){const p=document.createElement('pre');
p.textContent=typeof obj==='string'?obj:JSON.stringify(obj,null,2);e.appendChild(p)}
trace.appendChild(e);atBottom(trace)}
const es=new EventSource('/events');
es.onopen=()=>{conn.textContent='SSE 已连接';cdot.className='dot on'};
es.onerror=()=>{conn.textContent='SSE 断开,重连…';cdot.className='dot off'};
es.onmessage=e=>{const ev=JSON.parse(e.data);
if(ev.kind==='request'){
if(ev.endpoint)$('#endpoint').textContent=ev.url||ev.endpoint;
logEv('out','出站 · 已签名 POST',{url:ev.url,session_id:ev.session_id,'X-LB-Signature':ev.sig});
}else if(ev.kind==='ack'){
const id=ev.data&&ev.data.data&&ev.data.data.accepted_message_id;
sys(`LangBot 已接收 · HTTP ${ev.status}`);
logEv('ack','入站确认 202',{status:ev.status,accepted_message_id:id||'-'});
}else if(ev.kind==='reply'){
const sig=ev.sig_ok?'<span class=ok>验签通过</span>':'<span class=bad>验签失败</span>';
bubble('bot',ev.text,`seq=${ev.sequence} · ${ev.is_final?'<b>FINAL</b>':'中间段'} · ${sig}`);
logEv('reply',`回调 · seq ${ev.sequence}${ev.is_final?' · FINAL':''}`,
{session_id:ev.session_id,sequence:ev.sequence,is_final:ev.is_final,sig_ok:ev.sig_ok,text:ev.text});
}};
async function send(){
const t=inp.value.trim();if(!t)return;inp.value='';btn.disabled=true;
bubble('me',t,'已签名 → POST /bots/&lt;uuid&gt;');
try{await fetch('/send',{method:'POST',headers:{'Content-Type':'application/json'},
body:JSON.stringify({session_id:sidIn.value.trim()||'playground-1',text:t})});}
catch(e){sys('发送失败:'+e)}
btn.disabled=false;inp.focus();}
btn.onclick=send;inp.addEventListener('keydown',e=>{if(e.key==='Enter')send()});
$('#reset').onclick=()=>{sidIn.value='playground-'+Math.random().toString(36).slice(2,7);
sys('已切换到新会话 '+sidIn.value);};
sys('调试台就绪 · 每条消息都会真实发往运行中的 http_bot,右侧可观察签名 / 202 / 回调全过程。');
</script>
</body></html>"""
async def main():
api_key, bot_uuid = db_lookup()
callback_url = f'http://{PUBLIC_IP}:{PORT}/callback'
print(f'[init] http_bot uuid = {bot_uuid}')
print(f'[init] callback_url = {callback_url}')
ok = await configure_bot(api_key, bot_uuid, callback_url)
if not ok:
print('[warn] bot config update failed; check the API key / payload shape')
app = web.Application()
app['bot_uuid'] = bot_uuid
app.router.add_get('/', index)
app.router.add_post('/send', send)
app.router.add_post('/callback', callback)
app.router.add_get('/events', events)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, '0.0.0.0', PORT)
await site.start()
print(f'\n ▶ 打开: http://{PUBLIC_IP}:{PORT}/\n')
while True:
await asyncio.sleep(3600)
if __name__ == '__main__':
asyncio.run(main())
-48
View File
@@ -1,48 +0,0 @@
# Page Bot Adapter — Embed Demo
> English | [中文](./README.zh.md)
A single self-contained HTML page that demos the LangBot **Page Bot**
(`web_page_bot`) embeddable chat widget — the one you drop onto any website with
a single `<script>` tag.
Full guide: [docs.langbot.app — Page Bot](https://docs.langbot.app/en/usage/platforms/webpage).
## Files
| File | What it is |
|---|---|
| `index.html` | **Browser demo** — open it, point it at a running LangBot instance + a Page Bot you created, and it loads the live embed widget so you can chat with the bot exactly as a site visitor would. Zero deps, no build step. |
## How to use
1. In the LangBot WebUI, create a bot with the **Page Bot** (`页面机器人`)
adapter and bind it to a working pipeline. Copy its **bot UUID** from the
generated embed code.
2. Open `index.html` in a browser. Any of these work:
- double-click the file, or
- serve the folder: `python3 -m http.server 8930` then open
`http://localhost:8930/examples/web-page-bot/`.
3. Fill in:
- **LangBot base URL** — where your instance is reachable from the browser
(e.g. `http://localhost:5300`, or your public address).
- **Page Bot UUID** — from step 1.
- **Widget title** — optional, sets the `data-title` attribute.
4. Click **Load widget**. A floating chat bubble appears in the bottom-right
corner — click it and chat.
The page also renders the exact `<script>` snippet you'd paste into your own
site (before `</body>`), and updates it live as you edit the fields.
## What it demonstrates
- The embed contract: `<script data-title="…" src="<base>/api/v1/embed/<uuid>/widget.js"></script>`.
- `widget.js` is served by LangBot pre-configured for that bot UUID — title,
bubble icon, language and optional Cloudflare Turnstile protection all come
from the bot's config, no page changes needed.
- Messages travel over a WebSocket to the bot's bound pipeline; replies stream
back into the bubble.
> The widget loads `widget.js` from your LangBot instance, so the **base URL
> must be reachable from the browser** you open this page in. If LangBot runs on
> a server, use its public address instead of `localhost`.
-44
View File
@@ -1,44 +0,0 @@
# 页面机器人适配器 —— 嵌入演示
> [English](./README.md) | 中文
一个自包含的单文件 HTML 页面,用于演示 LangBot **页面机器人**
(`web_page_bot`) 的可嵌入聊天组件 —— 也就是你用一行 `<script>` 标签就能放到任意
网站上的那个组件。
完整指南:[docs.langbot.app —— 页面机器人](https://docs.langbot.app/zh/usage/platforms/webpage)。
## 文件清单
| 文件 | 是什么 |
|---|---|
| `index.html` | **浏览器演示页** —— 打开它,填上一个运行中的 LangBot 实例地址 + 你创建的页面机器人,它就会加载真实的嵌入组件,让你像网站访客一样和机器人对话。零依赖,无需构建。 |
## 使用方法
1. 在 LangBot WebUI 中,用 **页面机器人**`web_page_bot`)适配器创建一个机器人,
并绑定一个可用的流水线。从生成的嵌入代码里复制它的 **机器人 UUID**
2. 在浏览器中打开 `index.html`,以下任一方式皆可:
- 直接双击该文件;或
- 起一个静态服务:`python3 -m http.server 8930`,然后打开
`http://localhost:8930/examples/web-page-bot/`
3. 填写:
- **LangBot base URL** —— 你的实例在该浏览器中可访问的地址
(例如 `http://localhost:5300`,或你的公网地址)。
- **页面机器人 UUID** —— 第 1 步里复制的。
- **组件标题** —— 可选,对应 `data-title` 属性。
4. 点击 **Load widget**。页面右下角会出现一个浮动聊天气泡 —— 点开即可对话。
页面还会实时渲染出你需要粘贴到自己网站(放在 `</body>` 前)的那段 `<script>`
代码,并随着你编辑输入框同步更新。
## 它演示了什么
- 嵌入契约:`<script data-title="…" src="<base>/api/v1/embed/<uuid>/widget.js"></script>`
- `widget.js` 由 LangBot 针对该机器人 UUID 预配置后下发 —— 标题、气泡图标、语言
以及可选的 Cloudflare Turnstile 防护,全部来自机器人配置,无需改动页面。
- 消息通过 WebSocket 发往机器人绑定的流水线,回复流式回到气泡中。
> 组件会从你的 LangBot 实例加载 `widget.js`,因此 **base URL 必须能从你打开本页
> 的浏览器访问到**。如果 LangBot 部署在服务器上,请用它的公网地址而非
> `localhost`
-205
View File
@@ -1,205 +0,0 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>LangBot Page Bot · Embed Demo</title>
<style>
:root {
--bg: #f7f8fa; --panel: #ffffff; --line: #e8eaed; --ink: #1f2329;
--mut: #8a909a; --brand: #2563eb; --brand-soft: #eef3ff;
--ok: #16a34a; --bad: #dc2626; --code: #f3f4f6;
}
* { box-sizing: border-box; }
html, body { height: 100%; }
body {
margin: 0; background: var(--bg); color: var(--ink);
font: 14px/1.6 -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto,
"PingFang SC", "Microsoft YaHei", sans-serif;
}
.top {
height: 52px; background: var(--panel); border-bottom: 1px solid var(--line);
display: flex; align-items: center; gap: 10px; padding: 0 18px;
}
.logo {
width: 26px; height: 26px; border-radius: 7px; background: var(--brand);
display: grid; place-items: center; color: #fff; font-weight: 700; font-size: 14px;
}
.top b { font-size: 15px; }
.top .ver { font-size: 12px; color: var(--mut); }
.wrap { max-width: 760px; margin: 0 auto; padding: 28px 18px 80px; }
.hero h1 { margin: 8px 0 6px; font-size: 22px; }
.hero p { margin: 0 0 4px; color: var(--mut); }
.card {
background: var(--panel); border: 1px solid var(--line); border-radius: 12px;
padding: 20px; margin-top: 20px;
}
.card h3 {
margin: 0 0 14px; font-size: 14px; font-weight: 600; color: #4b5563;
display: flex; align-items: center; gap: 8px;
}
.card h3 .num {
width: 20px; height: 20px; border-radius: 50%; background: var(--brand-soft);
color: var(--brand); display: grid; place-items: center; font-size: 12px; font-weight: 700;
}
.field { margin-bottom: 14px; }
.field:last-child { margin-bottom: 0; }
.field label { display: block; font-size: 12px; color: var(--mut); margin-bottom: 5px; }
.field input {
width: 100%; border: 1px solid var(--line); border-radius: 9px;
padding: 10px 12px; font-size: 14px; outline: none; font-family: inherit;
}
.field input:focus { border-color: var(--brand); box-shadow: 0 0 0 3px var(--brand-soft); }
.hint { font-size: 12px; color: var(--mut); margin-top: 5px; }
.hint code { background: var(--code); border-radius: 5px; padding: 1px 5px; font-size: 11px; }
.actions { display: flex; gap: 10px; margin-top: 18px; align-items: center; }
button {
border: 0; border-radius: 9px; padding: 10px 18px; font-size: 14px;
font-weight: 500; cursor: pointer; font-family: inherit;
}
.btn-primary { background: var(--brand); color: #fff; }
.btn-primary:disabled { opacity: .5; cursor: default; }
.btn-ghost { background: #fff; border: 1px solid var(--line); color: #4b5563; }
.status { font-size: 13px; color: var(--mut); }
.status .ok { color: var(--ok); }
.status .bad { color: var(--bad); }
pre {
background: #0f172a; color: #e2e8f0; border-radius: 10px; padding: 14px 16px;
overflow: auto; font: 12px/1.6 ui-monospace, SFMono-Regular, Menlo, monospace;
margin: 0;
}
.snippet-row { position: relative; }
.snippet-row .copy {
position: absolute; top: 10px; right: 10px; background: rgba(255,255,255,.12);
color: #fff; border: 0; border-radius: 7px; padding: 5px 10px; font-size: 12px; cursor: pointer;
}
ul.steps { margin: 0; padding-left: 18px; color: #4b5563; }
ul.steps li { margin-bottom: 6px; }
</style>
</head>
<body>
<div class="top">
<div class="logo">L</div>
<b>Page Bot · Embed Demo</b>
<span class="ver">examples/web-page-bot</span>
</div>
<div class="wrap">
<div class="hero">
<h1>Try the LangBot Page Bot widget</h1>
<p>Point this page at a running LangBot instance and a <strong>Page Bot</strong> you created,</p>
<p>then load the live embed widget below to chat with it — exactly as your site visitors would.</p>
</div>
<div class="card">
<h3><span class="num">1</span> Connect your Page Bot</h3>
<div class="field">
<label for="base">LangBot base URL</label>
<input id="base" placeholder="http://localhost:5300" value="http://localhost:5300" />
<div class="hint">The address where your LangBot instance is reachable from this browser. No trailing slash.</div>
</div>
<div class="field">
<label for="uuid">Page Bot UUID</label>
<input id="uuid" placeholder="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" />
<div class="hint">Create a bot with the <code>Page Bot</code> adapter in the WebUI, then copy its UUID from the embed code.</div>
</div>
<div class="field">
<label for="title">Widget title (optional)</label>
<input id="title" placeholder="LangBot" value="LangBot" />
</div>
<div class="actions">
<button id="load" class="btn-primary">Load widget</button>
<button id="unload" class="btn-ghost">Remove widget</button>
<span class="status" id="status">Not loaded.</span>
</div>
</div>
<div class="card">
<h3><span class="num">2</span> The embed snippet</h3>
<p style="margin:0 0 12px;color:var(--mut)">This is exactly what you paste into your own site (before <code>&lt;/body&gt;</code>). It updates as you edit the fields above.</p>
<div class="snippet-row">
<button class="copy" id="copy">Copy</button>
<pre id="snippet">&lt;script data-title="LangBot" src="http://localhost:5300/api/v1/embed/&lt;bot-uuid&gt;/widget.js"&gt;&lt;/script&gt;</pre>
</div>
</div>
<div class="card">
<h3><span class="num">3</span> How it works</h3>
<ul class="steps">
<li>The <code>&lt;script&gt;</code> tag pulls <code>widget.js</code> from your LangBot instance, pre-configured for that bot UUID.</li>
<li>A floating chat bubble appears in the bottom-right corner of the page.</li>
<li>Messages travel over a WebSocket to the bot's bound pipeline; replies stream back into the bubble.</li>
<li>Title, bubble icon, language and optional Cloudflare Turnstile protection are all set in the bot's config — no page changes needed.</li>
</ul>
</div>
</div>
<script>
var $ = function (s) { return document.querySelector(s); };
var baseEl = $("#base"), uuidEl = $("#uuid"), titleEl = $("#title"),
statusEl = $("#status"), snippetEl = $("#snippet");
var WIDGET_ID = "langbot-embed-demo-script";
function clean(v) { return (v || "").trim().replace(/\/+$/, ""); }
function buildSrc() {
var base = clean(baseEl.value) || "http://localhost:5300";
var uuid = uuidEl.value.trim() || "<bot-uuid>";
return base + "/api/v1/embed/" + uuid + "/widget.js";
}
function refreshSnippet() {
var title = titleEl.value.trim() || "LangBot";
var src = buildSrc();
snippetEl.textContent =
'<script data-title="' + title + '" src="' + src + '"><\/script>';
}
function setStatus(html) { statusEl.innerHTML = html; }
function removeWidget() {
var old = document.getElementById(WIDGET_ID);
if (old) old.remove();
// The widget injects its own DOM (bubble + panel). Clear the common containers it creates.
document.querySelectorAll('[id^="langbot-"]').forEach(function (n) {
if (n.id !== WIDGET_ID) n.remove();
});
}
function loadWidget() {
var uuid = uuidEl.value.trim();
var uuidRe = /^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i;
if (!uuidRe.test(uuid)) {
setStatus('<span class="bad">Enter a valid bot UUID first.</span>');
return;
}
removeWidget();
var s = document.createElement("script");
s.id = WIDGET_ID;
s.setAttribute("data-title", titleEl.value.trim() || "LangBot");
s.src = buildSrc();
s.onload = function () {
setStatus('<span class="ok">Widget loaded — look bottom-right.</span>');
};
s.onerror = function () {
setStatus('<span class="bad">Failed to load widget.js — check the base URL and that the bot is enabled.</span>');
};
document.body.appendChild(s);
setStatus("Loading…");
}
$("#load").onclick = loadWidget;
$("#unload").onclick = function () {
removeWidget();
setStatus("Widget removed.");
};
$("#copy").onclick = function () {
navigator.clipboard.writeText(snippetEl.textContent).then(function () {
var b = $("#copy"); b.textContent = "Copied"; setTimeout(function () { b.textContent = "Copy"; }, 1200);
});
};
[baseEl, uuidEl, titleEl].forEach(function (el) { el.addEventListener("input", refreshSnippet); });
refreshSnippet();
</script>
</body>
</html>
+2 -2
View File
@@ -1,6 +1,6 @@
[project]
name = "langbot"
version = "4.10.5"
version = "4.10.2"
description = "Production-grade platform for building agentic IM bots"
readme = "README.md"
license-files = ["LICENSE"]
@@ -70,7 +70,7 @@ dependencies = [
"chromadb>=1.0.0,<2.0.0",
"qdrant-client (>=1.15.1,<2.0.0)",
"pyseekdb==1.1.0.post3",
"langbot-plugin==0.4.12",
"langbot-plugin==0.4.5",
"asyncpg>=0.30.0",
"line-bot-sdk>=3.19.0",
"matrix-nio>=0.25.2",
Executable → Regular
View File
View File
View File
View File
View File
View File
View File
View File
View File
View File
View File
View File
View File
View File
View File
View File
-32
View File
@@ -42,38 +42,6 @@ MyPlugin/
Each component has a `.yaml` (metadata) and `.py` (implementation).
## README & i18n convention (enforced on the marketplace)
A plugin published to LangBot Space serves a localized README on its detail page.
The resolver (`langbot-space` `PluginService.GetPluginREADME`) works like this:
- **Root `README.md` MUST be in English.** It is the default and the fallback —
when no per-language README matches the viewer's locale, the page serves the
root `README.md`. A non-English root README makes the English/default view show
the wrong language.
- **All other languages live under `readme/README_{lang}.md`** — e.g.
`readme/README_zh_Hans.md`, `readme/README_ja_JP.md`. The 8 supported locales:
`en_US, zh_Hans, zh_Hant, ja_JP, th_TH, vi_VN, es_ES, ru_RU`.
- `manifest.yaml` `metadata.label` / `metadata.description` should carry the same
8-locale i18n set (`repository` must be a real, alive URL).
```
MyPlugin/
├── manifest.yaml
├── README.md # English (default + fallback) — REQUIRED, must be English
└── readme/
├── README_zh_Hans.md
├── README_zh_Hant.md
├── README_ja_JP.md
├── README_th_TH.md
├── README_vi_VN.md
├── README_es_ES.md
└── README_ru_RU.md
```
`manifest.yaml` (incl. `repository`) is the source of truth — the marketplace
syncs from it, so edit the package and re-publish rather than patching live data.
## Critical SDK Pitfalls
### 1. MessageChain is a RootModel — iterate directly
View File
View File
View File
View File
View File
View File
View File
+1 -3
View File
@@ -1,5 +1,3 @@
"""LangBot - Production-grade platform for building agentic IM bots"""
from importlib.metadata import version
__version__ = version('langbot')
__version__ = '4.10.2'
@@ -1,9 +1,6 @@
from __future__ import annotations
from langbot.pkg.utils import constants
from .. import group
from .box_visibility import should_hide_box_runtime_status
@group.group_class('box', '/api/v1/box')
@@ -12,7 +9,6 @@ class BoxRouterGroup(group.RouterGroup):
@self.route('/status', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
async def _() -> str:
status = await self.ap.box_service.get_status()
status['hidden'] = should_hide_box_runtime_status(constants.edition, status.get('enabled'))
return self.success(data=status)
@self.route('/sessions', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
@@ -1,5 +0,0 @@
from __future__ import annotations
def should_hide_box_runtime_status(edition: str, box_enabled: bool | None) -> bool:
return edition == 'cloud' and box_enabled is False
@@ -138,39 +138,6 @@ class MonitoringRouterGroup(group.RouterGroup):
}
)
@self.route('/tool-calls', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
async def get_tool_calls() -> str:
"""Get tool call records"""
bot_ids = quart.request.args.getlist('botId')
pipeline_ids = quart.request.args.getlist('pipelineId')
session_ids = quart.request.args.getlist('sessionId')
start_time_str = quart.request.args.get('startTime')
end_time_str = quart.request.args.get('endTime')
limit = int(quart.request.args.get('limit', 100))
offset = int(quart.request.args.get('offset', 0))
start_time = parse_iso_datetime(start_time_str)
end_time = parse_iso_datetime(end_time_str)
tool_calls, total = await self.ap.monitoring_service.get_tool_calls(
bot_ids=bot_ids if bot_ids else None,
pipeline_ids=pipeline_ids if pipeline_ids else None,
session_ids=session_ids if session_ids else None,
start_time=start_time,
end_time=end_time,
limit=limit,
offset=offset,
)
return self.success(
data={
'tool_calls': tool_calls,
'total': total,
'limit': limit,
'offset': offset,
}
)
@self.route('/embedding-calls', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
async def get_embedding_calls() -> str:
"""Get embedding call records"""
@@ -317,16 +284,6 @@ class MonitoringRouterGroup(group.RouterGroup):
offset=0,
)
# Get tool calls
tool_calls, tool_calls_total = await self.ap.monitoring_service.get_tool_calls(
bot_ids=bot_ids if bot_ids else None,
pipeline_ids=pipeline_ids if pipeline_ids else None,
start_time=start_time,
end_time=end_time,
limit=limit,
offset=0,
)
# Get sessions
sessions, sessions_total = await self.ap.monitoring_service.get_sessions(
bot_ids=bot_ids if bot_ids else None,
@@ -361,14 +318,12 @@ class MonitoringRouterGroup(group.RouterGroup):
'overview': overview,
'messages': messages,
'llmCalls': llm_calls,
'toolCalls': tool_calls,
'embeddingCalls': embedding_calls,
'sessions': sessions,
'errors': errors,
'totalCount': {
'messages': messages_total,
'llmCalls': llm_calls_total,
'toolCalls': tool_calls_total,
'embeddingCalls': embedding_calls_total,
'sessions': sessions_total,
'errors': errors_total,
@@ -86,10 +86,6 @@ 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,
}
@@ -103,8 +99,6 @@ 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,
@@ -114,8 +108,6 @@ 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()
@@ -18,6 +18,7 @@ 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')
@@ -36,21 +37,30 @@ 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"')
@@ -62,29 +72,3 @@ 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()
@@ -2,7 +2,6 @@ from __future__ import annotations
import quart
import traceback
from urllib.parse import unquote
from ... import group
@@ -29,11 +28,11 @@ class MCPRouterGroup(group.RouterGroup):
traceback.print_exc()
return self.http_status(500, -1, f'Failed to create MCP server: {str(e)}')
@self.route(
'/servers/<path:server_name>', methods=['GET', 'PUT', 'DELETE'], auth_type=group.AuthType.USER_TOKEN
)
@self.route('/servers/<server_name>', methods=['GET', 'PUT', 'DELETE'], auth_type=group.AuthType.USER_TOKEN)
async def _(server_name: str) -> str:
"""获取、更新或删除MCP服务器配置"""
from urllib.parse import unquote
server_name = unquote(server_name)
server_data = await self.ap.mcp_service.get_mcp_server_by_name(server_name)
@@ -58,72 +57,12 @@ class MCPRouterGroup(group.RouterGroup):
except Exception as e:
return self.http_status(500, -1, f'Failed to delete MCP server: {str(e)}')
@self.route('/servers/<path:server_name>/test', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
@self.route('/servers/<server_name>/test', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
async def _(server_name: str) -> str:
"""测试MCP服务器连接"""
from urllib.parse import unquote
server_name = unquote(server_name)
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/<path: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/<path: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/<path:server_name>/logs', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
async def _(server_name: str) -> str:
"""Get logs from an MCP server"""
server_name = unquote(server_name)
try:
limit = int(quart.request.args.get('limit', 200))
except (TypeError, ValueError):
limit = 200
limit = min(limit, 500)
level = quart.request.args.get('level') or None
logs = await self.ap.mcp_service.get_mcp_server_logs(server_name, limit=limit, level=level)
return self.success(data={'logs': logs})
@self.route('/servers/<path: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)}')
@@ -1,7 +1,5 @@
from __future__ import annotations
import quart
from ... import group
@@ -11,41 +9,25 @@ class ToolsRouterGroup(group.RouterGroup):
@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
tools = await self.ap.tool_mgr.get_all_tools()
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')
tool_list = []
for tool in tools:
tool_list.append(
{
'name': tool.name,
'description': tool.description,
'human_desc': tool.human_desc,
'parameters': tool.parameters,
}
)
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,
)
}
)
return self.success(data={'tools': tool_list})
@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(include_skill_authoring=True)
tools = await self.ap.tool_mgr.get_all_tools()
for tool in tools:
if tool.name == tool_name:
@@ -1,5 +1,3 @@
import base64
import quart
from .. import group
@@ -32,50 +30,6 @@ 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."""
@@ -195,13 +195,6 @@ 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')
-32
View File
@@ -199,35 +199,3 @@ 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,
)
)
@@ -243,7 +243,6 @@ class MaintenanceService:
tables = {
'messages': persistence_monitoring.MonitoringMessage.id,
'llm_calls': persistence_monitoring.MonitoringLLMCall.id,
'tool_calls': persistence_monitoring.MonitoringToolCall.id,
'embedding_calls': persistence_monitoring.MonitoringEmbeddingCall.id,
'errors': persistence_monitoring.MonitoringError.id,
'sessions': persistence_monitoring.MonitoringSession.session_id,
+2 -67
View File
@@ -48,17 +48,6 @@ class MCPService:
if total_extensions >= max_extensions:
raise ValueError(f'Maximum number of extensions ({max_extensions}) reached')
server_name = str(server_data.get('name') or '').strip()
if not server_name:
raise ValueError('MCP server name is required')
server_data['name'] = server_name
existing_result = await self.ap.persistence_mgr.execute_async(
sqlalchemy.select(persistence_mcp.MCPServer).where(persistence_mcp.MCPServer.name == server_name)
)
if existing_result.first() is not None:
raise ValueError(f'MCP server already exists: {server_name}')
server_data['uuid'] = str(uuid.uuid4())
await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_mcp.MCPServer).values(server_data))
@@ -147,32 +136,6 @@ 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"""
@@ -188,22 +151,10 @@ class MCPService:
persisted_session = runtime_mcp_session
async def _refresh_and_report() -> None:
# Testing a persisted server should REUSE its live shared-session
# process, not rebuild it. Try a lightweight refresh (a real
# list_tools probe over the existing connection) first; only fall
# back to a full start() when the session has no live connection
# to probe (never connected, or the process is actually gone).
needs_start = persisted_session.status == MCPSessionStatus.ERROR or persisted_session.session is None
if needs_start:
if persisted_session.status == MCPSessionStatus.ERROR:
await persisted_session.start()
else:
try:
await persisted_session.refresh()
except Exception:
# The live connection was stale/dropped: reconnect once
# (reusing the live managed process where possible) and
# re-probe, instead of reporting a false failure.
await persisted_session.start()
await persisted_session.refresh()
# Surface the discovered tools so the config page can render them
# even for an already-hosted server.
ctx.metadata['runtime_info'] = persisted_session.get_runtime_info_dict()
@@ -244,19 +195,3 @@ class MCPService:
context=ctx,
)
return wrapper.id
async def get_mcp_server_logs(self, server_name: str, limit: int = 200, level: str | None = None) -> list[dict]:
"""Get recent log lines captured from the MCP server's stderr."""
session = self.ap.tool_mgr.mcp_tool_loader.get_session(server_name)
if not session:
return []
# Get logs from the session's buffer
logs = list(session._log_buffer)
# Filter by level if specified
if level:
logs = [log for log in logs if log.get('level') == level]
# Return the most recent 'limit' logs
return logs[-limit:]
@@ -2,7 +2,6 @@ from __future__ import annotations
import uuid
import datetime
import json
import sqlalchemy
from ....core import app
@@ -51,12 +50,6 @@ class MonitoringService:
persistence_monitoring.MonitoringLLMCall.timestamp,
persistence_monitoring.MonitoringLLMCall.id,
),
(
'monitoring_tool_calls',
persistence_monitoring.MonitoringToolCall,
persistence_monitoring.MonitoringToolCall.timestamp,
persistence_monitoring.MonitoringToolCall.id,
),
(
'monitoring_embedding_calls',
persistence_monitoring.MonitoringEmbeddingCall,
@@ -138,68 +131,6 @@ class MonitoringService:
await autocommit_conn.execute(sqlalchemy.text('PRAGMA wal_checkpoint(TRUNCATE)'))
await autocommit_conn.execute(sqlalchemy.text('VACUUM'))
def _serialize_tool_payload(self, payload: object, max_length: int = 20000) -> str | None:
"""Serialize tool arguments/results for monitoring storage."""
if payload is None:
return None
if isinstance(payload, str):
text = payload
else:
try:
text = json.dumps(payload, ensure_ascii=False, default=str)
except Exception:
text = str(payload)
if len(text) <= max_length:
return text
return f'{text[:max_length]}... [truncated {len(text) - max_length} chars]'
async def _get_message_for_tool_context(
self,
message_id: str | None = None,
session_id: str | None = None,
):
if message_id:
result = await self.ap.persistence_mgr.execute_async(
sqlalchemy.select(persistence_monitoring.MonitoringMessage).where(
persistence_monitoring.MonitoringMessage.id == message_id
)
)
row = result.first()
if row:
return row[0]
if not session_id:
return None
user_query = (
sqlalchemy.select(persistence_monitoring.MonitoringMessage)
.where(
sqlalchemy.and_(
persistence_monitoring.MonitoringMessage.session_id == session_id,
persistence_monitoring.MonitoringMessage.role == 'user',
)
)
.order_by(persistence_monitoring.MonitoringMessage.timestamp.desc())
.limit(1)
)
result = await self.ap.persistence_mgr.execute_async(user_query)
row = result.first()
if row:
return row[0]
any_query = (
sqlalchemy.select(persistence_monitoring.MonitoringMessage)
.where(persistence_monitoring.MonitoringMessage.session_id == session_id)
.order_by(persistence_monitoring.MonitoringMessage.timestamp.desc())
.limit(1)
)
result = await self.ap.persistence_mgr.execute_async(any_query)
row = result.first()
return row[0] if row else None
# ========== Recording Methods ==========
async def record_message(
@@ -289,57 +220,6 @@ class MonitoringService:
return call_id
async def record_tool_call(
self,
tool_name: str,
tool_source: str,
duration: int,
status: str = 'success',
bot_id: str | None = None,
bot_name: str | None = None,
pipeline_id: str | None = None,
pipeline_name: str | None = None,
session_id: str | None = None,
message_id: str | None = None,
arguments: object | None = None,
result: object | None = None,
error_message: str | None = None,
) -> str:
"""Record a tool call."""
context_message = await self._get_message_for_tool_context(message_id=message_id, session_id=session_id)
if context_message:
bot_id = bot_id or context_message.bot_id
bot_name = bot_name or context_message.bot_name
pipeline_id = pipeline_id or context_message.pipeline_id
pipeline_name = pipeline_name or context_message.pipeline_name
session_id = session_id or context_message.session_id
message_id = message_id or context_message.id
call_id = str(uuid.uuid4())
call_data = {
'id': call_id,
'timestamp': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None),
'tool_name': tool_name,
'tool_source': tool_source,
'duration': max(0, duration),
'status': status,
'bot_id': bot_id or 'unknown',
'bot_name': bot_name or 'Unknown',
'pipeline_id': pipeline_id or 'unknown',
'pipeline_name': pipeline_name or 'Unknown',
'session_id': session_id,
'message_id': message_id,
'arguments': self._serialize_tool_payload(arguments),
'result': self._serialize_tool_payload(result),
'error_message': self._serialize_tool_payload(error_message),
}
await self.ap.persistence_mgr.execute_async(
sqlalchemy.insert(persistence_monitoring.MonitoringToolCall).values(call_data)
)
return call_id
async def record_embedding_call(
self,
model_name: str,
@@ -869,58 +749,6 @@ class MonitoringService:
total,
)
async def get_tool_calls(
self,
bot_ids: list[str] | None = None,
pipeline_ids: list[str] | None = None,
session_ids: list[str] | None = None,
start_time: datetime.datetime | None = None,
end_time: datetime.datetime | None = None,
limit: int = 100,
offset: int = 0,
) -> tuple[list[dict], int]:
"""Get tool calls with filters"""
conditions = []
if bot_ids:
conditions.append(persistence_monitoring.MonitoringToolCall.bot_id.in_(bot_ids))
if pipeline_ids:
conditions.append(persistence_monitoring.MonitoringToolCall.pipeline_id.in_(pipeline_ids))
if session_ids:
conditions.append(persistence_monitoring.MonitoringToolCall.session_id.in_(session_ids))
if start_time:
conditions.append(persistence_monitoring.MonitoringToolCall.timestamp >= start_time)
if end_time:
conditions.append(persistence_monitoring.MonitoringToolCall.timestamp <= end_time)
count_query = sqlalchemy.select(sqlalchemy.func.count(persistence_monitoring.MonitoringToolCall.id))
if conditions:
count_query = count_query.where(sqlalchemy.and_(*conditions))
count_result = await self.ap.persistence_mgr.execute_async(count_query)
total = count_result.scalar() or 0
query = sqlalchemy.select(persistence_monitoring.MonitoringToolCall).order_by(
persistence_monitoring.MonitoringToolCall.timestamp.desc()
)
if conditions:
query = query.where(sqlalchemy.and_(*conditions))
query = query.limit(limit).offset(offset)
result = await self.ap.persistence_mgr.execute_async(query)
tool_calls_rows = result.all()
return (
[
self.ap.persistence_mgr.serialize_model(
persistence_monitoring.MonitoringToolCall, row[0] if isinstance(row, tuple) else row
)
for row in tool_calls_rows
],
total,
)
async def get_embedding_calls(
self,
start_time: datetime.datetime | None = None,
@@ -1143,34 +971,6 @@ class MonitoringService:
else:
error_llm_calls += 1
# Get tool calls for this session
tool_query = (
sqlalchemy.select(persistence_monitoring.MonitoringToolCall)
.where(persistence_monitoring.MonitoringToolCall.session_id == session_id)
.order_by(persistence_monitoring.MonitoringToolCall.timestamp.asc())
)
tool_result = await self.ap.persistence_mgr.execute_async(tool_query)
tool_rows = tool_result.all()
tool_calls = [
self.ap.persistence_mgr.serialize_model(
persistence_monitoring.MonitoringToolCall, row[0] if isinstance(row, tuple) else row
)
for row in tool_rows
]
total_tool_calls = len(tool_rows)
success_tool_calls = 0
error_tool_calls = 0
total_tool_duration = 0
for row in tool_rows:
tool_call = row[0] if isinstance(row, tuple) else row
total_tool_duration += tool_call.duration
if tool_call.status == 'success':
success_tool_calls += 1
else:
error_tool_calls += 1
# Get errors for this session
error_query = (
sqlalchemy.select(persistence_monitoring.MonitoringError)
@@ -1214,14 +1014,6 @@ class MonitoringService:
'total_tokens': total_tokens,
'average_duration_ms': int(total_duration / total_llm_calls) if total_llm_calls > 0 else 0,
},
'tool_calls': tool_calls,
'tool_stats': {
'total_calls': total_tool_calls,
'success_calls': success_tool_calls,
'error_calls': error_tool_calls,
'total_duration_ms': total_tool_duration,
'average_duration_ms': int(total_tool_duration / total_tool_calls) if total_tool_calls > 0 else 0,
},
'errors': errors,
'session_duration_seconds': session_duration_seconds,
}
@@ -100,8 +100,6 @@ 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(
@@ -195,8 +193,6 @@ class PipelineService:
'enable_all_mcp_servers': True,
'plugins': [],
'mcp_servers': [],
'mcp_resources': [],
'mcp_resource_agent_read_enabled': True,
}
),
}
@@ -221,8 +217,6 @@ 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
@@ -242,14 +236,10 @@ 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)
+16 -16
View File
@@ -20,15 +20,6 @@ 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))
@@ -37,7 +28,9 @@ class UserService:
return result_list is not None and len(result_list) > 0
async def create_user(self, user_email: str, password: str) -> None:
hashed_password = await self._hash_password(password)
ph = argon2.PasswordHasher()
hashed_password = ph.hash(password)
await self.ap.persistence_mgr.execute_async(
sqlalchemy.insert(user.User).values(user=user_email, password=hashed_password, account_type='local')
@@ -76,7 +69,9 @@ class UserService:
if not user_obj.password:
raise ValueError('请使用 Space 账户登录')
await self._verify_password(user_obj.password, password)
ph = argon2.PasswordHasher()
ph.verify(user_obj.password, password)
return await self.generate_jwt_token(user_email)
@@ -98,13 +93,17 @@ class UserService:
return jwt.decode(token, jwt_secret, algorithms=['HS256'])['user']
async def reset_password(self, user_email: str, new_password: str) -> None:
hashed_password = await self._hash_password(new_password)
ph = argon2.PasswordHasher()
hashed_password = ph.hash(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')
@@ -112,9 +111,9 @@ class UserService:
if not user_obj.password:
raise ValueError('No local password set, please set a password first')
await self._verify_password(user_obj.password, current_password)
ph.verify(user_obj.password, current_password)
hashed_password = await self._hash_password(new_password)
hashed_password = ph.hash(new_password)
await self.ap.persistence_mgr.execute_async(
sqlalchemy.update(user.User).where(user.User.user == user_email).values(password=hashed_password)
@@ -233,6 +232,7 @@ 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')
await self._verify_password(user_obj.password, current_password)
ph.verify(user_obj.password, current_password)
hashed_password = await self._hash_password(new_password)
hashed_password = ph.hash(new_password)
await self.ap.persistence_mgr.execute_async(
sqlalchemy.update(user.User).where(user.User.user == user_email).values(password=hashed_password)
)
+2 -5
View File
@@ -82,6 +82,7 @@ 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
@@ -98,7 +99,6 @@ 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(
@@ -1152,9 +1152,6 @@ 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
@@ -1179,7 +1176,7 @@ class BoxService:
return
host_path = os.path.realpath(spec.host_path)
if self.shares_filesystem_with_box and not os.path.isdir(host_path):
if 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:
+1 -11
View File
@@ -84,17 +84,7 @@ class CommandManager:
privilege = 1
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:
if f'{query.launcher_type.value}_{query.launcher_id}' in self.ap.instance_config.data['admins']:
privilege = 2
ctx = command_context.ExecuteContext(
-14
View File
@@ -3,20 +3,6 @@ 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"""
@@ -49,28 +49,6 @@ class MonitoringLLMCall(Base):
message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True) # Associated message ID
class MonitoringToolCall(Base):
"""Tool call records"""
__tablename__ = 'monitoring_tool_calls'
id = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True)
timestamp = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, index=True)
tool_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
tool_source = sqlalchemy.Column(sqlalchemy.String(50), nullable=False) # native, plugin, mcp, skill
duration = sqlalchemy.Column(sqlalchemy.Integer, nullable=False) # milliseconds
status = sqlalchemy.Column(sqlalchemy.String(50), nullable=False) # success, error
bot_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True)
bot_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
pipeline_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True)
pipeline_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
session_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)
message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)
arguments = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
result = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
error_message = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
class MonitoringSession(Base):
"""Session tracking records"""
@@ -26,14 +26,7 @@ class LegacyPipeline(Base):
extensions_preferences = sqlalchemy.Column(
sqlalchemy.JSON,
nullable=False,
default={
'enable_all_plugins': True,
'enable_all_mcp_servers': True,
'plugins': [],
'mcp_servers': [],
'mcp_resources': [],
'mcp_resource_agent_read_enabled': True,
},
default={'enable_all_plugins': True, 'enable_all_mcp_servers': True, 'plugins': [], 'mcp_servers': []},
)
@@ -1,84 +0,0 @@
"""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')
@@ -1,95 +0,0 @@
"""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)
@@ -1,17 +0,0 @@
from langbot.pkg.entity.persistence import monitoring as persistence_monitoring
from .. import migration
@migration.migration_class(26)
class DBMigrateMonitoringToolCalls(migration.DBMigration):
"""Add monitoring_tool_calls table"""
async def upgrade(self):
"""Upgrade"""
async with self.ap.persistence_mgr.get_db_engine().begin() as conn:
await conn.run_sync(persistence_monitoring.MonitoringToolCall.__table__.create, checkfirst=True)
async def downgrade(self):
"""Downgrade"""
async with self.ap.persistence_mgr.get_db_engine().begin() as conn:
await conn.run_sync(persistence_monitoring.MonitoringToolCall.__table__.drop, checkfirst=True)
@@ -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.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'
session_id = f'{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.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'
session_id = f'{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.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'
session_id = f'{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.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'
session_id = f'{query.launcher_type}_{query.launcher_id}'
await ap.monitoring_service.record_llm_call(
bot_id=bot_id,
+1 -12
View File
@@ -96,15 +96,6 @@ class RuntimePipeline:
extensions_prefs = pipeline_entity.extensions_preferences or {}
self.enable_all_plugins = extensions_prefs.get('enable_all_plugins', True)
self.enable_all_mcp_servers = extensions_prefs.get('enable_all_mcp_servers', True)
local_agent_config = (pipeline_entity.config or {}).get('ai', {}).get('local-agent', {})
self.mcp_resource_attachments = local_agent_config.get(
'mcp-resources',
extensions_prefs.get('mcp_resources', []),
)
self.mcp_resource_agent_read_enabled = 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
@@ -125,8 +116,6 @@ 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:
@@ -189,7 +178,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.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'
session_id = f'{query.launcher_type}_{query.launcher_id}'
# Update message status to error
if message_id:
@@ -1,274 +0,0 @@
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(),
},
}
+3 -25
View File
@@ -25,21 +25,6 @@ class PreProcessor(stage.PipelineStage):
- use_funcs
"""
@staticmethod
def _filter_selected_tools(
tools: list,
local_agent_config: dict,
) -> list:
if local_agent_config.get('enable-all-tools', True) is not False:
return tools
selected_tools = local_agent_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,
@@ -47,7 +32,6 @@ class PreProcessor(stage.PipelineStage):
) -> entities.StageProcessResult:
"""Process"""
selected_runner = query.pipeline_config['ai']['runner']['runner']
local_agent_config = query.pipeline_config.get('ai', {}).get('local-agent', {})
include_skill_authoring = (
selected_runner == 'local-agent' and getattr(self.ap, 'skill_service', None) is not None
)
@@ -59,7 +43,7 @@ class PreProcessor(stage.PipelineStage):
if selected_runner == 'local-agent':
# Read model config — new format is { primary: str, fallbacks: [str] },
# but handle legacy plain string for backward compatibility
model_config = local_agent_config.get('model', {})
model_config = query.pipeline_config['ai']['local-agent'].get('model', {})
if isinstance(model_config, str):
# Legacy format: plain UUID string
primary_uuid = model_config
@@ -129,14 +113,11 @@ class PreProcessor(stage.PipelineStage):
# Get bound plugins and MCP servers for filtering tools
bound_plugins = query.variables.get('_pipeline_bound_plugins', None)
bound_mcp_servers = query.variables.get('_pipeline_bound_mcp_servers', None)
include_mcp_resource_tools = query.variables.get('_pipeline_mcp_resource_agent_read_enabled', True)
all_tools = await self.ap.tool_mgr.get_all_tools(
query.use_funcs = await self.ap.tool_mgr.get_all_tools(
bound_plugins,
bound_mcp_servers,
include_skill_authoring=include_skill_authoring,
include_mcp_resource_tools=include_mcp_resource_tools,
)
query.use_funcs = self._filter_selected_tools(all_tools, local_agent_config)
self.ap.logger.debug(f'Bound plugins: {bound_plugins}')
self.ap.logger.debug(f'Bound MCP servers: {bound_mcp_servers}')
@@ -147,14 +128,11 @@ class PreProcessor(stage.PipelineStage):
if not query.use_funcs and query.variables.get('_fallback_model_uuids'):
bound_plugins = query.variables.get('_pipeline_bound_plugins', None)
bound_mcp_servers = query.variables.get('_pipeline_bound_mcp_servers', None)
include_mcp_resource_tools = query.variables.get('_pipeline_mcp_resource_agent_read_enabled', True)
all_tools = await self.ap.tool_mgr.get_all_tools(
query.use_funcs = await self.ap.tool_mgr.get_all_tools(
bound_plugins,
bound_mcp_servers,
include_skill_authoring=include_skill_authoring,
include_mcp_resource_tools=include_mcp_resource_tools,
)
query.use_funcs = self._filter_selected_tools(all_tools, local_agent_config)
sender_name = ''
@@ -9,7 +9,6 @@ from datetime import datetime
from .. import handler
from ... import entities
from ... import plugin_diagnostics
from ....provider import runner as runner_module
import langbot_plugin.api.entities.events as events
@@ -59,13 +58,6 @@ 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)
@@ -4,7 +4,6 @@ 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
@@ -53,13 +52,6 @@ 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)
+14 -30
View File
@@ -9,7 +9,6 @@ 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
@@ -40,35 +39,20 @@ class SendResponseBackStage(stage.PipelineStage):
has_chunks = any(isinstance(msg, provider_message.MessageChunk) for msg in query.resp_messages)
# TODO 命令与流式的兼容性问题
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,
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,
)
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)
@@ -3,7 +3,6 @@ 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
@@ -79,11 +78,6 @@ 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)
@@ -135,10 +129,8 @@ 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.
@@ -146,13 +138,6 @@ 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,
@@ -195,12 +180,6 @@ 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(
+12 -125
View File
@@ -4,7 +4,6 @@ import asyncio
import traceback
import datetime
import json
import time
import aiocqhttp
import pydantic
@@ -17,37 +16,6 @@ from ...utils import image
import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger
_GROUP_NAME_CACHE_TTL_SECONDS = 3600
_GROUP_NAME_NEGATIVE_CACHE_TTL_SECONDS = 60
_GROUP_NAME_LOOKUP_TIMEOUT_SECONDS = 2
_GROUP_MEMBER_INFO_CACHE_TTL_SECONDS = 86400
_GROUP_MEMBER_INFO_NEGATIVE_CACHE_TTL_SECONDS = 600
_GROUP_MEMBER_INFO_LOOKUP_TIMEOUT_SECONDS = 2
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
def _get_field(data: dict, key: str, default: str = '') -> str:
value = data.get(key)
if value is None:
return default
return str(value)
def _get_group_member_name(sender: dict) -> str:
return _get_field(sender, 'card') or _get_field(sender, 'nickname') or _get_field(sender, 'user_id')
def _get_group_name_placeholder(group_id: typing.Union[int, str]) -> str:
return f'Group {group_id}'
class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
@staticmethod
async def yiri2target(
@@ -67,7 +35,7 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
elif type(msg) is platform_message.Image:
arg = ''
if msg.base64:
arg = _normalize_base64_payload(msg.base64)
arg = msg.base64
msg_list.append(aiocqhttp.MessageSegment.image(f'base64://{arg}'))
elif msg.url:
arg = msg.url
@@ -82,7 +50,7 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
elif type(msg) is platform_message.Voice:
arg = ''
if msg.base64:
arg = _normalize_base64_payload(msg.base64)
arg = msg.base64
msg_list.append(aiocqhttp.MessageSegment.record(f'base64://{arg}'))
elif msg.url:
arg = msg.url
@@ -94,10 +62,7 @@ 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):
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}})
msg_list.append({'type': 'file', 'data': {'file': msg.url, 'name': msg.name}})
elif isinstance(msg, platform_message.Face):
if msg.face_type == 'face':
msg_list.append(aiocqhttp.MessageSegment.face(msg.face_id))
@@ -359,96 +324,16 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
def __init__(self):
self._group_name_cache: dict[typing.Union[int, str], tuple[str, float]] = {}
self._group_name_negative_cache: dict[typing.Union[int, str], float] = {}
self._group_member_info_cache: dict[
tuple[typing.Union[int, str], typing.Union[int, str]], tuple[dict, float]
] = {}
self._group_member_info_negative_cache: dict[tuple[typing.Union[int, str], typing.Union[int, str]], float] = {}
@staticmethod
async def yiri2target(event: platform_events.MessageEvent, bot_account_id: int):
return event.source_platform_object
async def _get_group_name(self, group_id: typing.Union[int, str], bot=None) -> str:
now = time.monotonic()
if group_id in self._group_name_cache:
group_name, expires_at = self._group_name_cache[group_id]
if expires_at > now:
return group_name
del self._group_name_cache[group_id]
if group_id in self._group_name_negative_cache:
expires_at = self._group_name_negative_cache[group_id]
if expires_at > now:
return ''
del self._group_name_negative_cache[group_id]
if bot is None:
return ''
try:
group_info = await asyncio.wait_for(
bot.get_group_info(group_id=group_id),
timeout=_GROUP_NAME_LOOKUP_TIMEOUT_SECONDS,
)
except Exception:
self._group_name_negative_cache[group_id] = now + _GROUP_NAME_NEGATIVE_CACHE_TTL_SECONDS
return ''
group_name = _get_field(group_info, 'group_name') if isinstance(group_info, dict) else ''
if group_name:
self._group_name_cache[group_id] = (group_name, now + _GROUP_NAME_CACHE_TTL_SECONDS)
self._group_name_negative_cache.pop(group_id, None)
else:
self._group_name_negative_cache[group_id] = now + _GROUP_NAME_NEGATIVE_CACHE_TTL_SECONDS
return group_name
async def _get_group_member_info(
self,
group_id: typing.Union[int, str],
user_id: typing.Union[int, str],
bot=None,
) -> dict:
now = time.monotonic()
cache_key = (group_id, user_id)
if cache_key in self._group_member_info_cache:
member_info, expires_at = self._group_member_info_cache[cache_key]
if expires_at > now:
return member_info
del self._group_member_info_cache[cache_key]
if cache_key in self._group_member_info_negative_cache:
expires_at = self._group_member_info_negative_cache[cache_key]
if expires_at > now:
return {}
del self._group_member_info_negative_cache[cache_key]
if bot is None:
return {}
try:
member_info = await asyncio.wait_for(
bot.get_group_member_info(group_id=group_id, user_id=user_id),
timeout=_GROUP_MEMBER_INFO_LOOKUP_TIMEOUT_SECONDS,
)
except Exception:
self._group_member_info_negative_cache[cache_key] = now + _GROUP_MEMBER_INFO_NEGATIVE_CACHE_TTL_SECONDS
return {}
if isinstance(member_info, dict) and member_info:
self._group_member_info_cache[cache_key] = (
member_info,
now + _GROUP_MEMBER_INFO_CACHE_TTL_SECONDS,
)
self._group_member_info_negative_cache.pop(cache_key, None)
return member_info
self._group_member_info_negative_cache[cache_key] = now + _GROUP_MEMBER_INFO_NEGATIVE_CACHE_TTL_SECONDS
return {}
async def target2yiri(self, event: aiocqhttp.Event, bot=None):
@staticmethod
async def target2yiri(event: aiocqhttp.Event, bot=None):
yiri_chain = await AiocqhttpMessageConverter.target2yiri(event.message, event.message_id, bot)
if event.message_type == 'group':
permission = 'MEMBER'
group_name = await self._get_group_name(event.group_id, bot) or _get_group_name_placeholder(event.group_id)
special_title = _get_field(event.sender, 'title')
if not special_title:
member_info = await self._get_group_member_info(event.group_id, event.sender['user_id'], bot)
special_title = _get_field(member_info, 'title')
if 'role' in event.sender:
if event.sender['role'] == 'admin':
@@ -458,14 +343,14 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
converted_event = platform_events.GroupMessage(
sender=platform_entities.GroupMember(
id=event.sender['user_id'], # message_seq 放哪?
member_name=_get_group_member_name(event.sender),
member_name=event.sender['nickname'],
permission=permission,
group=platform_entities.Group(
id=event.group_id,
name=group_name,
name=event.sender['nickname'],
permission=platform_entities.Permission.Member,
),
special_title=special_title,
special_title=event.sender['title'] if 'title' in event.sender else '',
),
message_chain=yiri_chain,
time=event.time,
@@ -489,7 +374,7 @@ class AiocqhttpAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
bot: aiocqhttp.CQHttp = pydantic.Field(exclude=True, default_factory=aiocqhttp.CQHttp)
message_converter: AiocqhttpMessageConverter = AiocqhttpMessageConverter()
event_converter: AiocqhttpEventConverter = pydantic.Field(default_factory=AiocqhttpEventConverter)
event_converter: AiocqhttpEventConverter = AiocqhttpEventConverter()
on_websocket_connection_event_cache: typing.List[typing.Callable[[aiocqhttp.Event], None]] = []
@@ -548,7 +433,9 @@ class AiocqhttpAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
elif isinstance(component, platform_message.Image):
img_data = {}
if component.base64:
b64 = _normalize_base64_payload(component.base64)
b64 = component.base64
if b64.startswith('data:'):
b64 = b64.split(',', 1)[-1] if ',' in b64 else b64
img_data['file'] = f'base64://{b64}'
elif component.url:
img_data['file'] = component.url
@@ -1,509 +0,0 @@
"""HTTP Bot adapter — standalone server-to-server platform adapter.
Lets any external backend drive a LangBot pipeline over plain HTTP:
* **Inbound** the backend POSTs a signed message to the unified webhook
route ``POST /bots/<bot_uuid>``; this adapter verifies the signature, builds
a platform event carrying the caller-defined ``session_id`` as the launcher
id, and fires it into the normal pipeline (so message aggregation, N->1,
works for free).
* **Outbound** every ``reply_message`` / ``reply_message_chunk`` the pipeline
emits is delivered as a signed POST to the configured ``callback_url``. A
single turn may emit many replies (1->M); each is one callback, ordered per
session via a small worker queue.
Design notes:
* The callback URL is taken **only** from adapter config (never from the
inbound message) to keep the SSRF surface closed.
* Replies for one ``session_id`` are delivered in ``sequence`` order; the
caller knows a turn is complete when ``is_final: true`` arrives.
* No new HTTP route is registered the existing unified webhook dispatcher
(``pkg/api/http/controller/groups/webhooks.py``) calls
``handle_unified_webhook`` on this adapter.
See docs/platforms/http-bot.md for the full integration guide.
"""
from __future__ import annotations
import asyncio
import json
import time
import typing
import uuid
from datetime import datetime
import aiohttp
import pydantic
import quart
import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
import langbot_plugin.api.entities.builtin.platform.message as platform_message
import langbot_plugin.api.entities.builtin.platform.events as platform_events
import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger
from . import http_bot_signing as signing
from ...utils import httpclient
# Error envelope codes (HTTP status -> body code), documented in the design doc.
_ERR = {
'bad_request': (400, 40001),
'bad_signature': (401, 40101),
'duplicate': (409, 40901),
'too_large': (413, 41301),
'internal': (500, 50001),
}
# Max accepted inbound body size (bytes).
_MAX_BODY = 1 * 1024 * 1024
# Idempotency dedup window (seconds) and cap.
_IDEMPOTENCY_TTL = 600
_IDEMPOTENCY_MAX = 4096
class _SessionOutbound:
"""Per-session outbound state: ordered delivery queue + sequence counter."""
def __init__(self) -> None:
self.queue: asyncio.Queue = asyncio.Queue(maxsize=1000)
self.worker: asyncio.Task | None = None
self.sequence: int = 0
self.last_was_final: bool = True # so the first reply of a turn starts at seq 1
class _SyncCollector:
"""Collects reply parts for a /sync request and resolves when the turn ends."""
def __init__(self) -> None:
self.parts: list = []
self.done: asyncio.Event = asyncio.Event()
class HttpBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
"""Standalone HTTP adapter (inbound webhook + outbound callbacks)."""
bot_uuid: str = pydantic.Field(default='', exclude=True)
listeners: dict[
typing.Type[platform_events.Event],
typing.Callable[[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None],
] = pydantic.Field(default_factory=dict, exclude=True)
# session_id -> outbound state
outbound_states: dict[str, _SessionOutbound] = pydantic.Field(default_factory=dict, exclude=True)
# idempotency key -> accepted-at epoch
idempotency_cache: dict[str, float] = pydantic.Field(default_factory=dict, exclude=True)
# session_id -> sync collector (set while a /sync request is awaiting a turn)
sync_waiters: dict[str, '_SyncCollector'] = pydantic.Field(default_factory=dict, exclude=True)
model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger, **kwargs):
super().__init__(config=config, logger=logger, **kwargs)
self.bot_account_id = 'http_bot'
self.outbound_states = {}
self.idempotency_cache = {}
self.sync_waiters = {}
# -- framework hooks ------------------------------------------------------
def set_bot_uuid(self, bot_uuid: str) -> None:
"""Called by the bot manager so the adapter knows its own bot uuid."""
object.__setattr__(self, 'bot_uuid', bot_uuid)
def get_launcher_id(self, event: platform_events.MessageEvent) -> str:
"""Map an inbound event to a LangBot launcher id.
We return the caller-defined ``session_id`` (stashed on the sender /
group id at inbound time) so that each external session maps 1:1 to an
isolated LangBot session.
"""
if isinstance(event, platform_events.GroupMessage):
return str(event.sender.group.id)
return str(event.sender.id)
def register_listener(
self,
event_type: typing.Type[platform_events.Event],
func: typing.Callable[
[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], typing.Awaitable[None]
],
):
self.listeners[event_type] = func
def unregister_listener(
self,
event_type: typing.Type[platform_events.Event],
func: typing.Callable[
[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], typing.Awaitable[None]
],
):
self.listeners.pop(event_type, None)
async def is_muted(self, group_id: int) -> bool:
return False
async def is_stream_output_supported(self) -> bool:
return True
async def run_async(self):
# Purely webhook-driven; nothing to poll. Stay alive.
while True:
await asyncio.sleep(3600)
async def kill(self):
# Cancel any outbound workers.
for state in self.outbound_states.values():
if state.worker and not state.worker.done():
state.worker.cancel()
return True
# -- inbound --------------------------------------------------------------
def _err(self, kind: str, detail: str = ''):
status, code = _ERR[kind]
return quart.jsonify({'code': code, 'msg': detail or kind, 'data': None}), status
def _prune_idempotency(self) -> None:
now = time.time()
if len(self.idempotency_cache) > _IDEMPOTENCY_MAX:
self.idempotency_cache.clear()
return
expired = [k for k, ts in self.idempotency_cache.items() if now - ts > _IDEMPOTENCY_TTL]
for k in expired:
self.idempotency_cache.pop(k, None)
async def handle_unified_webhook(self, bot_uuid: str, path: str, request):
"""Handle an inbound POST from the unified webhook dispatcher.
Sub-path routing:
(no path) -> push a message
"reset" -> reset a session's conversation (body: {session_id, session_type?})
"sync" -> push a message and wait for the final reply (collapses 1->M)
"""
object.__setattr__(self, 'bot_uuid', bot_uuid)
if path == 'reset':
return await self._handle_reset(request)
if path == 'sync':
return await self._handle_inbound(request, sync=True)
if path in ('', None):
return await self._handle_inbound(request, sync=False)
return self._err('bad_request', f'unknown sub-path: {path}')
async def _read_and_verify(self, request) -> tuple[dict | None, typing.Any]:
"""Read body, enforce size + signature. Returns (data, error_response)."""
body = await request.get_data()
if body and len(body) > _MAX_BODY:
return None, self._err('too_large', 'message too large')
if self.config.get('signature_required', True):
ok, reason = signing.verify(
secret=self.config.get('inbound_secret', ''),
body=body,
timestamp=request.headers.get(signing.HEADER_TIMESTAMP),
signature=request.headers.get(signing.HEADER_SIGNATURE),
)
if not ok:
await self.logger.warning(f'http_bot inbound signature rejected: {reason}')
return None, self._err('bad_signature', f'invalid signature: {reason}')
try:
data = json.loads(body)
except (json.JSONDecodeError, ValueError):
return None, self._err('bad_request', 'body is not valid JSON')
if not isinstance(data, dict):
return None, self._err('bad_request', 'body must be a JSON object')
return data, None
def _build_event(self, data: dict) -> tuple[platform_events.MessageEvent, str, str, str]:
"""Build a platform event from inbound data.
Returns (event, session_id, session_type, message_id).
"""
session_id = str(data['session_id'])
session_type = data.get('session_type') or self.config.get('default_session_type', 'person')
sender_meta = data.get('sender') or {}
sender_name = str(sender_meta.get('name', 'User'))
message_id = 'in_' + uuid.uuid4().hex
chain = platform_message.MessageChain.model_validate(data['message'])
# Carry the inbound message id + timestamp as the Source component.
chain.insert(0, platform_message.Source(id=message_id, time=datetime.now()))
if session_type == 'group':
group = platform_entities.Group(
id=session_id,
name=str(sender_meta.get('group_name', session_id)),
permission=platform_entities.Permission.Member,
)
sender = platform_entities.GroupMember(
id=str(sender_meta.get('id', session_id)),
member_name=sender_name,
group=group,
permission=platform_entities.Permission.Member,
)
event = platform_events.GroupMessage(sender=sender, message_chain=chain, time=datetime.now().timestamp())
else:
sender = platform_entities.Friend(id=session_id, nickname=sender_name, remark=sender_name)
event = platform_events.FriendMessage(sender=sender, message_chain=chain, time=datetime.now().timestamp())
return event, session_id, session_type, message_id
async def _handle_inbound(self, request, sync: bool):
data, err = await self._read_and_verify(request)
if err is not None:
return err
if 'session_id' not in data or 'message' not in data:
return self._err('bad_request', 'session_id and message are required')
# Idempotency.
idem = request.headers.get(signing.HEADER_IDEMPOTENCY)
if idem:
self._prune_idempotency()
if idem in self.idempotency_cache:
return self._err('duplicate', 'idempotency key already accepted')
self.idempotency_cache[idem] = time.time()
try:
event, session_id, session_type, message_id = self._build_event(data)
except Exception as e: # noqa: BLE001
return self._err('bad_request', f'failed to parse message: {e}')
listener = self.listeners.get(type(event))
if listener is None:
return self._err('internal', 'no listener registered for event type')
if sync:
return await self._run_sync(event, listener, session_id, message_id)
# Fire-and-collect: kick the pipeline, return 202 immediately.
asyncio.create_task(listener(event, self))
return quart.jsonify(
{
'code': 0,
'msg': 'accepted',
'data': {
'session_id': session_id,
'accepted_message_id': message_id,
'aggregating': True,
},
}
), 202
async def _handle_reset(self, request):
data, err = await self._read_and_verify(request)
if err is not None:
return err
if 'session_id' not in data:
return self._err('bad_request', 'session_id is required')
session_id = str(data['session_id'])
session_type = data.get('session_type') or self.config.get('default_session_type', 'person')
launcher_type = 'group' if session_type == 'group' else 'person'
removed = await self._reset_session(launcher_type, session_id)
return quart.jsonify({'code': 0, 'msg': 'reset', 'data': {'session_id': session_id, 'removed': removed}}), 200
async def _reset_session(self, launcher_type: str, launcher_id: str) -> bool:
"""Drop the matching session so the next message starts a fresh conversation."""
sess_mgr = self.ap.sess_mgr
before = len(sess_mgr.session_list)
sess_mgr.session_list = [
s
for s in sess_mgr.session_list
if not (
str(s.launcher_type.value if hasattr(s.launcher_type, 'value') else s.launcher_type) == launcher_type
and str(s.launcher_id) == launcher_id
)
]
return len(sess_mgr.session_list) < before
# -- outbound -------------------------------------------------------------
@staticmethod
def _extract_session_id(message_source: platform_events.MessageEvent) -> str:
if isinstance(message_source, platform_events.GroupMessage):
return str(message_source.sender.group.id)
return str(message_source.sender.id)
@staticmethod
def _extract_reply_to(message_source: platform_events.MessageEvent) -> str:
for comp in message_source.message_chain:
if isinstance(comp, platform_message.Source):
return str(comp.id)
return ''
def _next_sequence(self, session_id: str, is_final: bool) -> int:
state = self.outbound_states.setdefault(session_id, _SessionOutbound())
if state.last_was_final:
state.sequence = 1
else:
state.sequence += 1
state.last_was_final = is_final
return state.sequence
async def _enqueue_callback(self, session_id: str, payload: dict) -> None:
state = self.outbound_states.setdefault(session_id, _SessionOutbound())
if state.worker is None or state.worker.done():
state.worker = asyncio.create_task(self._outbound_worker(session_id, state))
try:
state.queue.put_nowait(payload)
except asyncio.QueueFull:
# Drop oldest to bound memory, then enqueue (best-effort, at-least-once).
try:
state.queue.get_nowait()
except asyncio.QueueEmpty:
pass
await self.logger.warning(f'http_bot outbound queue full for session {session_id}; dropped oldest')
state.queue.put_nowait(payload)
async def _outbound_worker(self, session_id: str, state: _SessionOutbound) -> None:
while True:
payload = await state.queue.get()
try:
await self._deliver_callback(payload)
except Exception as e: # noqa: BLE001
await self.logger.error(f'http_bot callback delivery failed for {session_id}: {e}')
finally:
state.queue.task_done()
async def _deliver_callback(self, payload: dict) -> None:
callback_url = self.config.get('callback_url', '')
if not callback_url:
await self.logger.warning('http_bot has no callback_url configured; dropping reply')
return
body = json.dumps(payload, ensure_ascii=False).encode()
secret = self.config.get('outbound_secret') or self.config.get('inbound_secret', '')
ts, sig = signing.sign(secret, body)
headers = {
'Content-Type': 'application/json',
signing.HEADER_TIMESTAMP: ts,
signing.HEADER_SIGNATURE: sig,
}
timeout = aiohttp.ClientTimeout(total=int(self.config.get('callback_timeout', 15)))
max_retries = int(self.config.get('callback_max_retries', 3))
session = httpclient.get_session()
attempt = 0
while True:
attempt += 1
try:
async with session.post(callback_url, data=body, headers=headers, timeout=timeout) as resp:
if resp.status < 400:
return
if resp.status < 500 or attempt > max_retries:
await self.logger.warning(f'http_bot callback {callback_url} -> {resp.status}, giving up')
return
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
if attempt > max_retries:
await self.logger.warning(f'http_bot callback {callback_url} failed after {attempt} tries: {e}')
return
await asyncio.sleep(min(2 ** (attempt - 1), 30))
async def _emit_reply(
self,
message_source: platform_events.MessageEvent,
message: platform_message.MessageChain,
is_final: bool,
stream: bool,
) -> dict:
session_id = self._extract_session_id(message_source)
reply_to = self._extract_reply_to(message_source)
sequence = self._next_sequence(session_id, is_final)
parts = [c.model_dump() if hasattr(c, 'model_dump') else c.__dict__ for c in message]
payload = {
'session_id': session_id,
'reply_to': reply_to,
'sequence': sequence,
'is_final': is_final,
'stream': stream,
'message': parts,
'timestamp': datetime.now().isoformat(),
}
# If a /sync request is awaiting this session, collect instead of POSTing.
collector = self.sync_waiters.get(session_id)
if collector is not None:
collector.parts.extend(parts)
if is_final:
collector.done.set()
return payload
await self._enqueue_callback(session_id, payload)
return payload
async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain) -> dict:
"""Proactively push a message to a session (target_id == session_id)."""
sequence = self._next_sequence(str(target_id), is_final=True)
payload = {
'session_id': str(target_id),
'reply_to': '',
'sequence': sequence,
'is_final': True,
'stream': False,
'message': [c.model_dump() if hasattr(c, 'model_dump') else c.__dict__ for c in message],
'timestamp': datetime.now().isoformat(),
}
await self._enqueue_callback(str(target_id), payload)
return payload
async def reply_message(
self,
message_source: platform_events.MessageEvent,
message: platform_message.MessageChain,
quote_origin: bool = False,
) -> dict:
return await self._emit_reply(message_source, message, is_final=True, stream=False)
async def reply_message_chunk(
self,
message_source: platform_events.MessageEvent,
bot_message,
message: platform_message.MessageChain,
quote_origin: bool = False,
is_final: bool = False,
) -> dict:
message_is_final = is_final and getattr(bot_message, 'tool_calls', None) is None
return await self._emit_reply(message_source, message, is_final=message_is_final, stream=True)
# -- sync convenience mode ------------------------------------------------
async def _run_sync(self, event, listener, session_id: str, message_id: str):
"""Push a message and wait for the final reply, collapsing 1->M parts.
Lossy by design (drops streaming/ordering nuance); documented as such.
Concurrency-safe: routing is via the per-session ``_sync_waiters``
registry that ``_emit_reply`` consults, not by patching methods.
"""
if session_id in self.sync_waiters:
return self._err('duplicate', 'a sync request is already in flight for this session')
collector = _SyncCollector()
self.sync_waiters[session_id] = collector
try:
asyncio.create_task(listener(event, self))
timeout = int(self.config.get('callback_timeout', 15)) * 4
try:
await asyncio.wait_for(collector.done.wait(), timeout=timeout)
except asyncio.TimeoutError:
await self.logger.warning(f'http_bot sync wait timed out for session {session_id}')
finally:
self.sync_waiters.pop(session_id, None)
return quart.jsonify(
{
'code': 0,
'msg': 'ok',
'data': {
'session_id': session_id,
'reply_to': message_id,
'message': collector.parts,
},
}
), 200
@@ -1,9 +0,0 @@
<svg width="800px" height="800px" viewBox="0 0 64 64" fill="none" xmlns="http://www.w3.org/2000/svg">
<rect x="2" y="2" width="60" height="60" rx="14" fill="#2563EB"/>
<g stroke="#FFFFFF" stroke-width="3.6" stroke-linecap="round" stroke-linejoin="round" fill="none">
<!-- </> code icon -->
<path d="M24 22 L14 32 L24 42"/>
<path d="M40 22 L50 32 L40 42"/>
<path d="M36 18 L28 46"/>
</g>
</svg>

Before

Width:  |  Height:  |  Size: 416 B

@@ -1,153 +0,0 @@
apiVersion: v1
kind: MessagePlatformAdapter
metadata:
name: http_bot
label:
en_US: HTTP Bot
zh_Hans: HTTP 通用接入
zh_Hant: HTTP 通用接入
ja_JP: HTTP ボット
description:
en_US: Integrate any backend over plain HTTP. Push messages in via a signed webhook, receive replies on a callback URL. Server-to-server, no long-lived connection. Preserves message aggregation (N->1) and multi-part replies (1->M).
zh_Hans: 通过 HTTP 接入任意后端系统。以签名 Webhook 推入消息,在回调地址接收回复。面向服务间集成,无需长连接。完整保留消息聚合(多条合一)与多段回复(一条问、多条回)能力。
zh_Hant: 透過 HTTP 接入任意後端系統。以簽名 Webhook 推入訊息,在回調地址接收回覆。面向服務間整合,無需長連線。完整保留訊息聚合(多條合一)與多段回覆(一條問、多條回)能力。
ja_JP: 任意のバックエンドを HTTP で接続。署名付き Webhook でメッセージを送信し、コールバック URL で返信を受信します。サーバー間連携、長時間接続不要。メッセージ集約(N→1)とマルチパート返信(1→M)に対応。
icon: http_bot.svg
spec:
categories:
- popular
- global
help_links:
zh: https://docs.langbot.app/zh/platforms/http-bot
en: https://docs.langbot.app/en/platforms/http-bot
ja: https://docs.langbot.app/ja/platforms/http-bot
config:
- name: webhook_url
label:
en_US: Inbound Webhook URL
zh_Hans: 入站 Webhook 地址
zh_Hant: 入站 Webhook 地址
ja_JP: 受信 Webhook URL
description:
en_US: Copy this URL. Your backend POSTs messages here (signed with the inbound secret).
zh_Hans: 复制此地址。你的后端将消息以签名方式 POST 到这里。
zh_Hant: 複製此地址。你的後端將訊息以簽名方式 POST 到這裡。
ja_JP: この URL をコピーしてください。バックエンドは署名付きでここにメッセージを POST します。
type: webhook-url
required: false
default: ""
- name: inbound_secret
label:
en_US: Inbound Signing Secret
zh_Hans: 入站签名密钥
zh_Hant: 入站簽名密鑰
ja_JP: 受信署名シークレット
description:
en_US: HMAC-SHA256 secret your backend uses to sign inbound requests. LangBot verifies every inbound POST with it.
zh_Hans: 你的后端用于对入站请求做 HMAC-SHA256 签名的密钥;LangBot 据此校验每个入站 POST。
zh_Hant: 你的後端用於對入站請求做 HMAC-SHA256 簽名的密鑰;LangBot 據此校驗每個入站 POST。
ja_JP: バックエンドが受信リクエストの署名に使う HMAC-SHA256 シークレット。LangBot は受信 POST ごとに検証します。
type: string
required: true
default: ""
- name: callback_url
label:
en_US: Outbound Callback URL
zh_Hans: 出站回调地址
zh_Hant: 出站回調地址
ja_JP: 送信コールバック URL
description:
en_US: Where LangBot POSTs replies. One turn may trigger multiple callbacks (1->M). For security the callback URL is taken ONLY from this config and cannot be overridden per-message.
zh_Hans: LangBot 将回复 POST 到此地址。一轮对话可能触发多次回调(一问多答)。出于安全考虑,回调地址只取自此配置,不允许逐条消息覆盖。
zh_Hant: LangBot 將回覆 POST 到此地址。一輪對話可能觸發多次回調(一問多答)。出於安全考慮,回調地址只取自此配置,不允許逐條訊息覆蓋。
ja_JP: LangBot が返信を POST する先。1 ターンで複数回のコールバック(1→M)が発生し得ます。セキュリティ上、コールバック URL はこの設定からのみ取得し、メッセージ単位で上書きできません。
type: string
required: true
default: ""
- name: outbound_secret
label:
en_US: Outbound Signing Secret
zh_Hans: 出站签名密钥
zh_Hant: 出站簽名密鑰
ja_JP: 送信署名シークレット
description:
en_US: HMAC-SHA256 secret LangBot uses to sign outbound callbacks so your receiver can verify them. Falls back to the inbound secret when empty.
zh_Hans: LangBot 用于对出站回调签名的密钥,供你的接收端校验。留空时回退使用入站密钥。
zh_Hant: LangBot 用於對出站回調簽名的密鑰,供你的接收端校驗。留空時回退使用入站密鑰。
ja_JP: LangBot が送信コールバックの署名に使う HMAC-SHA256 シークレット。受信側で検証できます。空の場合は受信シークレットを使用します。
type: string
required: false
default: ""
- name: default_session_type
label:
en_US: Default Session Type
zh_Hans: 默认会话类型
zh_Hant: 預設會話類型
ja_JP: デフォルトセッションタイプ
description:
en_US: Session type used when an inbound message omits session_type.
zh_Hans: 入站消息未携带 session_type 时使用的会话类型。
zh_Hant: 入站訊息未攜帶 session_type 時使用的會話類型。
ja_JP: 受信メッセージに session_type がない場合に使用するセッションタイプ。
type: select
options:
- name: person
label:
en_US: Person (1-on-1)
zh_Hans: 个人(一对一)
zh_Hant: 個人(一對一)
ja_JP: 個人(1 対 1
- name: group
label:
en_US: Group
zh_Hans: 群组
zh_Hant: 群組
ja_JP: グループ
required: false
default: person
- name: signature_required
label:
en_US: Require Inbound Signature
zh_Hans: 强制入站签名校验
zh_Hant: 強制入站簽名校驗
ja_JP: 受信署名を必須にする
description:
en_US: When enabled (recommended), every inbound POST must carry a valid signature. Disable ONLY for local development behind a trusted network.
zh_Hans: 开启(推荐)后,每个入站 POST 都必须带有效签名。仅在受信任内网的本地开发时关闭。
zh_Hant: 開啟(推薦)後,每個入站 POST 都必須帶有效簽名。僅在受信任內網的本地開發時關閉。
ja_JP: 有効(推奨)にすると、すべての受信 POST に有効な署名が必要です。信頼できるネットワーク内のローカル開発時のみ無効化してください。
type: boolean
required: false
default: true
- name: callback_timeout
label:
en_US: Callback Timeout (seconds)
zh_Hans: 回调超时(秒)
zh_Hant: 回調逾時(秒)
ja_JP: コールバックタイムアウト(秒)
description:
en_US: Per-callback HTTP timeout.
zh_Hans: 单次回调的 HTTP 超时时间。
zh_Hant: 單次回調的 HTTP 逾時時間。
ja_JP: コールバックごとの HTTP タイムアウト。
type: integer
required: false
default: 15
- name: callback_max_retries
label:
en_US: Callback Max Retries
zh_Hans: 回调最大重试次数
zh_Hant: 回調最大重試次數
ja_JP: コールバック最大リトライ回数
description:
en_US: Retries on timeout or 5xx, with exponential backoff.
zh_Hans: 超时或 5xx 时按指数退避重试的次数。
zh_Hant: 逾時或 5xx 時按指數退避重試的次數。
ja_JP: タイムアウトまたは 5xx 時に指数バックオフでリトライする回数。
type: integer
required: false
default: 3
execution:
python:
path: ./http_bot.py
attr: HttpBotAdapter
@@ -1,95 +0,0 @@
"""HMAC signing utilities for the HTTP Bot adapter.
A dependency-free, symmetric HMAC-SHA256 scheme used in *both* directions:
signing_string = "{timestamp}." + raw_body_bytes
signature = "sha256=" + hex(HMAC_SHA256(secret, signing_string))
Inbound requests are signed by the caller and verified here; outbound
callbacks are signed here and verified by the caller. The scheme is trivial to
reproduce in any language (see docs/platforms/http-bot.md for JS/curl).
"""
from __future__ import annotations
import hashlib
import hmac
import time
# Header names (kept here so adapter + clients agree on a single source).
HEADER_TIMESTAMP = 'X-LB-Timestamp'
HEADER_SIGNATURE = 'X-LB-Signature'
HEADER_IDEMPOTENCY = 'X-LB-Idempotency-Key'
# Maximum allowed clock skew between signer and verifier (seconds).
DEFAULT_REPLAY_WINDOW = 300
def compute_signature(secret: str, body: bytes, timestamp: str | int) -> str:
"""Compute the ``sha256=<hex>`` signature for *body* at *timestamp*.
Args:
secret: Shared HMAC secret.
body: Raw request body bytes (exactly as sent on the wire).
timestamp: Unix timestamp (seconds) as str or int.
Returns:
The signature string, e.g. ``sha256=ab12...``.
"""
signing_string = f'{timestamp}.'.encode() + body
digest = hmac.new(secret.encode(), signing_string, hashlib.sha256).hexdigest()
return f'sha256={digest}'
def sign(secret: str, body: bytes, timestamp: int | None = None) -> tuple[str, str]:
"""Produce ``(timestamp, signature)`` for an outbound request.
Args:
secret: Shared HMAC secret.
body: Raw request body bytes.
timestamp: Optional fixed timestamp; defaults to ``int(time.time())``.
Returns:
``(timestamp_str, signature_str)``.
"""
ts = str(timestamp if timestamp is not None else int(time.time()))
return ts, compute_signature(secret, body, ts)
def verify(
secret: str,
body: bytes,
timestamp: str | None,
signature: str | None,
replay_window: int = DEFAULT_REPLAY_WINDOW,
) -> tuple[bool, str]:
"""Verify an inbound signature.
Args:
secret: Shared HMAC secret.
body: Raw request body bytes.
timestamp: Value of the timestamp header.
signature: Value of the signature header.
replay_window: Max allowed skew in seconds.
Returns:
``(ok, reason)``. ``reason`` is empty when ``ok`` is True, otherwise a
short machine-friendly cause (``missing_headers`` / ``bad_timestamp`` /
``expired`` / ``signature_mismatch``).
"""
if not timestamp or not signature:
return False, 'missing_headers'
try:
ts_int = int(float(timestamp))
except (ValueError, TypeError):
return False, 'bad_timestamp'
if abs(int(time.time()) - ts_int) > replay_window:
return False, 'expired'
expected = compute_signature(secret, body, timestamp)
if not hmac.compare_digest(expected, signature):
return False, 'signature_mismatch'
return True, ''
+1 -23
View File
@@ -2,7 +2,6 @@ from __future__ import annotations
import typing
import asyncio
import traceback
import uuid
import datetime
import pydantic
@@ -183,28 +182,7 @@ class WecomCSAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
)
async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):
if target_type != 'person':
raise ValueError('WeCom customer service only supports sending messages to person targets')
open_kfid = self.bot_account_id
external_userid = target_id
if '|' in target_id:
open_kfid, external_userid = target_id.split('|', 1)
if external_userid.startswith('u'):
external_userid = external_userid[1:]
if not open_kfid:
raise ValueError('WeCom customer service open_kfid is required before sending messages')
content_list = await WecomMessageConverter.yiri2target(message, self.bot)
for content in content_list:
msgid = f'langbot_{uuid.uuid4().hex}'
if content['type'] == 'text':
await self.bot.send_text_msg(
open_kfid=open_kfid,
external_userid=external_userid,
msgid=msgid,
content=content['content'],
)
pass
def set_bot_uuid(self, bot_uuid: str):
"""设置 bot UUID(用于生成 webhook URL"""
-14
View File
@@ -737,8 +737,6 @@ 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
@@ -747,21 +745,9 @@ 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 []
-21
View File
@@ -26,15 +26,6 @@ from ..core import app
from ..utils import constants
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.
@@ -932,18 +923,6 @@ 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(
@@ -417,30 +417,6 @@ class LocalAgentRunner(runner.RequestRunner):
ce.text = final_user_message_text
break
mcp_loader = getattr(getattr(self.ap, 'tool_mgr', None), 'mcp_tool_loader', None)
if mcp_loader is not None:
resource_context = await mcp_loader.build_resource_context_for_query(query)
if resource_context:
resource_addition = (
'\n\nMCP resource context selected by LangBot host:\n'
f'{resource_context}\n\n'
'Use this context as read-only reference material. If it conflicts with the user message, '
'ask for clarification before taking external actions.'
)
if isinstance(user_message.content, str):
user_message.content += resource_addition
elif isinstance(user_message.content, list):
appended = False
for ce in user_message.content:
if ce.type == 'text':
ce.text = (ce.text or '') + resource_addition
appended = True
break
if not appended:
user_message.content.append(
provider_message.ContentElement.from_text(resource_addition.strip())
)
req_messages = self._build_request_messages(query, user_message)
try:

Some files were not shown because too many files have changed in this diff Show More