包package
agentscope.appagentscope.app
module agentscope.app
agentscope.app 属于运行时与部署,服务化、调度、控制面与 Web 应用。 该包共收录 2 个公开类型。
The FastAPI based agent service module, which contains all service-related components and a configurable FastAPI app factory.
源码Source
agentscope/app/__init__.py:1子包Subpackages
类Classes
SubAgentTemplate类class | 子智能体模板,承载智能体的部署、调度与服务化运行。 A reusable blueprint for sub-agent creation within a team. |
函数Functions
create_app函数function | 创建 FastAPI 应用 Create and configure a FastAPI application. |
函数详情Function Details
create_app
def create_app(storage: StorageBase, message_bus: MessageBus, workspace_manager: WorkspaceManagerBase, knowledge_base_manager: KnowledgeBaseManagerBase | None = None, knowledge_parsers: list[ParserBase] | dict[str, ParserBase] | None = None, knowledge_chunkers: list[Type[ChunkerBase]] | None = None, blob_store: BlobStoreBase | None = None, enable_index_worker: bool = True, mcp_hubs: list[MCPHubBase] | None = None, skill_hubs: list[SkillHubBase] | None = None, *, enable_channel_worker: bool = True, enable_scheduler: bool = True, extra_credentials: list[Type[CredentialBase]] | None = None, extra_middlewares: list[FastAPIMiddleware] | None = None, extra_agent_middlewares: AgentMiddlewareFactory | None = None, extra_agent_tools: AgentToolFactory | None = None, custom_subagent_templates: list[SubAgentTemplate] | None = None, custom_agent_cls: Type[Agent] | None = None, resource_access_policy: ResourceAccessPolicyBase | None = None, channels: list[Type[ChannelBase]] | None = None, download_secret: str | None = None, realtime_ice_servers: list[dict[str, Any]] | None = None, title: str = 'AgentScope', version: str = __version__, **kwargs: Any) -> FastAPI
装配路由、中间件与运行时依赖,返回可直接启动的 FastAPI 实例。
This is the primary entry point for embedding AgentScope into an existing service or running it standalone.
参数Parameters
| 名称Name | 类型Type | 说明Description |
|---|---|---|
storage | StorageBase | 存储(类型 `StorageBase`)。The storage backend. Its lifecycle (`__aenter__` / `__aexit__`) is managed by the app lifespan. |
message_bus | MessageBus | 消息总线(类型 `MessageBus`)。The live message bus used for cross-session inbox delivery and idle-session triggers. Required — the bus is intentionally decoupled from `storage` so the persistence backend (e.g. SQL) can differ from the transport backend (Redis). Its lifecycle is also managed by the app lifespan. |
workspace_manager | WorkspaceManagerBase | 工作区管理器(类型 `WorkspaceManagerBase`)。The workspace manager. Required — every chat run and every `/workspace` endpoint depends on it. Its lifecycle ( `__aenter__` / `__aexit__`) is managed by the app lifespan. Pass a :class:`~agentscope.app._manager. LocalWorkspaceManager` for local-directory workspaces. |
knowledge_base_manager默认default None | KnowledgeBaseManagerBase | None | 知识库管理器(类型 `KnowledgeBaseManagerBase | None`)。The knowledge base manager that owns knowledge base lifecycle and serves :class:`~agentscope.rag.KnowledgeBase` runtime handles to both HTTP service and agent code. The manager carries its own vector store instance — its `__aenter__` / `__aexit__` enter and release that vector store, so the caller does not pass the vector store separately. `None` disables knowledge base endpoints entirely. |
knowledge_parsers默认default None | list[ParserBase] | dict[str, ParserBase] | None | 知识库 parsers(类型 `list[ParserBase] | dict[str, ParserBase] | None`)。Parsers registered for knowledge base document uploads. Pass a **list** to have the service route by each parser's `supported_media_types` (later entries override earlier ones for overlapping types, with a warning); pass a **dict** `media_type → parser` for explicit routing (one parser bound to multiple types, type aliases, ...). Defaults to `[TextParser()]` when `knowledge_base_manager` is set. |
knowledge_chunkers默认default None | list[Type[ChunkerBase]] | None | 知识库 chunkers(类型 `list[Type[ChunkerBase]] | None`)。The chunker classes users can choose from when creating a knowledge base. The chunker type and parameters are pinned on the knowledge base record and reconstructed by the index worker. Defaults to `[ApproxTokenChunker]` when `knowledge_base_manager` is set. |
blob_store默认default None | BlobStoreBase | None | 二进制对象存储(类型 `BlobStoreBase | None`)。Backend storing uploaded document bytes between the upload endpoint and the indexing worker. Required when `knowledge_base_manager` is set; defaults to :class:`~agentscope.app.rag.blob_store.LocalBlobStore` rooted at `./blobs`. Its lifecycle (`__aenter__` / `__aexit__`) is managed by the app lifespan. |
enable_index_worker默认default True | bool | 启用索引工作进程(类型 `bool`)。When `True` (embedded deployment) the API process starts an :class:`~agentscope.app._service.IndexWorker` and an :class:`~agentscope.app._service.IndexSweeper` in its lifespan, and dispatches indexing tasks via an in-process queue. When `False` (dedicated deployment) the API process performs no indexing — a separate worker process is expected to consume tasks from the message bus. No effect when `knowledge_base_manager` is `None`. |
mcp_hubs默认default None | list[MCPHubBase] | None | MCP hubs(类型 `list[MCPHubBase] | None`)。The MCP hubs that provide MCPs. |
skill_hubs默认default None | list[SkillHubBase] | None | 技能 hubs(类型 `list[SkillHubBase] | None`)。The SkillHubs that provide skills. |
enable_channel_workerkw默认default True | bool | 启用渠道工作进程(类型 `bool`)。Whether this process holds the channels' long connections. `True` (embedded deployment) suits a desktop build or a single API process. Set `False` when running dedicated channel workers: a platform gives one bot's events to one connection, so every replica connecting would either waste connections or duplicate messages. The channel API, the client factory and webhook delivery stay available either way — only the connections move. |
enable_schedulerkw默认default True | bool | 启用调度器(类型 `bool`)。Whether this process owns the schedule timers. APScheduler's jobstore is in-memory, so every process holding them fires every cron tick and a schedule runs once per replica — set `False` on all but one. The schedule API and the agent's schedule tools stay available either way; those processes persist the record and notify the owner over the bus. |
extra_credentialskw默认default None | list[Type[CredentialBase]] | None | extra 凭证(类型 `list[Type[CredentialBase]] | None`)。Additional :class:`~agentscope.credential.CredentialBase` subclasses to register before the app starts. Equivalent to calling :func:`~agentscope.credential.CredentialFactory. register_credential` for each class. |
extra_middlewareskw默认default None | list[FastAPIMiddleware] | None | extra middlewares(类型 `list[FastAPIMiddleware] | None`)。Additional ASGI middlewares to add to the application. |
extra_agent_middlewareskw默认default None | AgentMiddlewareFactory | None | extra 智能体 middlewares(类型 `AgentMiddlewareFactory | None`)。An async factory `(user_id, agent_id, session_id, workspace) -> awaitable of list[MiddlewareBase]` that produces extra :class:`~agentscope.middleware.MiddlewareBase` instances to attach to the agent on each invocation. Called once per agent assembly (i.e. per chat turn / scheduled trigger), so it can return user/session-specific middleware (auth, audit logging, tenant isolation, etc.). `workspace` is the session's resolved :class:`~agentscope.workspace.WorkspaceBase`, exposing `workdir` and `get_backend()` for filesystem-backed middleware such as :class:`~agentscope.middleware.AgenticMemoryMiddleware`. Factories written against the older three-argument signature keep working — the fourth argument is only passed to factories that accept it. The returned middlewares are appended to the framework-supplied ones (e.g. `ToolOffloadMiddleware`). |
extra_agent_toolskw默认default None | AgentToolFactory | None | extra 智能体工具(类型 `AgentToolFactory | None`)。An async factory `(user_id, agent_id, session_id) -> awaitable of list[ToolBase]` that produces extra :class:`~agentscope.tool.ToolBase` instances to register in the agent's toolkit on each invocation. Useful when tool availability depends on the caller (per-tenant integrations, user-specific credentials). The returned tools are added to the workspace-derived tools in the toolkit's `"basic"` group. |
custom_subagent_templateskw默认default None | list[SubAgentTemplate] | None | 自定义子智能体 templates(类型 `list[SubAgentTemplate] | None`)。Reusable blueprints for sub-agent creation within teams. Each template defines a sub-agent *type* (e.g. `"researcher"`, `"coder"`) with pre-configured system prompt, context config, ReAct config, permission context, and task context. When registered, the `AgentCreate` tool exposes a `subagent_type` parameter so the leader agent can route to the appropriate template. See :class:`~agentscope.app._types.SubAgentTemplate` for details. |
custom_agent_clskw默认default None | Type[Agent] | None | 自定义智能体 cls(类型 `Type[Agent] | None`)。A custom :class:`~agentscope.agent.Agent` subclass to use when assembling agents. When `None` (default), the built-in :class:`~agentscope.agent.Agent` is used. |
resource_access_policykw默认default None | ResourceAccessPolicyBase | None | 资源访问策略(类型 `ResourceAccessPolicyBase | None`)。Policy deciding whether a viewer may access credentials / agents / knowledge bases owned by another user. When `None` (default), a :class:`DenyAllResourceAccessPolicy` is installed which preserves the historical owner-isolated behavior. |
channelskw默认default None | list[Type[ChannelBase]] | None | 渠道(类型 `list[Type[ChannelBase]] | None`)。Channel adapter classes this service allows (e.g. `[DingTalkChannel, FeishuChannel, DiscordChannel]`). Each class self-describes its `channel_type`, credentials and config, so the service registers it without a separate table; pass a custom :class:`~agentscope.app.channel.ChannelBase` subclass to add a platform. When `None` (default), no channel types are registered and the channel feature stays off until the caller opts in by passing at least one adapter class. |
download_secretkw默认default None | str | None | 下载密钥(类型 `str | None`)。Signs the short-lived tokens that let a browser download a workspace file by navigation. Defaults to a value generated per process, which is fine for a single instance but **must be set explicitly behind a load balancer** — otherwise a token minted by one replica is rejected by the next, and downloads fail at random. |
realtime_ice_serverskw默认default None | list[dict[str, Any]] | None | 实时 ice servers(类型 `list[dict[str, Any]] | None`)。STUN and TURN servers exposed to browser WebRTC clients. When omitted, the JSON array in `AGENTSCOPE_REALTIME_ICE_SERVERS` is used. The default is an empty list, which permits host candidates without silently sending network metadata to a public STUN service. |
titlekw默认default 'AgentScope' | str | title(类型 `str`)。OpenAPI title shown in the docs UI. |
versionkw默认default __version__ | str | 版本(类型 `str`)。API version shown in the docs UI. |
kwargs | Any | 关键字参数(类型 `Any`)。The kwargs. |
FastAPI
返回FastAPI。`FastAPI`: A fully configured application ready to serve requests.
源码Source
agentscope/app/_app.py:113–478