类class agentscope.app.storage

StorageBase

class StorageBase(ABC)

源码Source agentscope/app/storage/_base.py:32–1447

存储抽象基类类。位于 agentscope.app.storage 包，继承自 ABC。核心成员包括 aclose、upsert_credential、list_credentials、get_credential、delete_credential、upsert_mcp。

The storage abstract base class. Declared in agentscope.app.storage. Extends ABC.

方法摘要Method Summary

方法Method	说明Description
`async def __aenter__() -> Self`async	进入异步上下文并返回自身。 Start the storage backend (open connection pool, etc.).
`async def __aexit__(exc_type: type[BaseException] \| None, exc_value: BaseException \| None, traceback: Any) -> None`async	退出异步上下文并清理资源。 Shut down the storage backend.
`async def aclose() -> None`async	返回 `aclose` 结果。 Release underlying connection resources. Default is a no-op.
`async def upsert_credential(user_id: str, credential_data:` `CredentialBase) -> str`asyncabstractabstractmethod	创建或更新凭证 in 存储。 Create or update a credential in the storage.
`async def list_credentials(user_id: str) -> list[CredentialRecord]`asyncabstractabstractmethod	列出凭证。 List all credentials for a given user.
`async def get_credential(user_id: str, credential_id: str) ->` `CredentialRecord \| None`asyncabstractabstractmethod	获取凭证。 Fetch a single credential record by id.
`async def delete_credential(user_id: str, credential_id: str) -> bool`asyncabstractabstractmethod	删除凭证。 Delete a credential.
`async def upsert_mcp(user_id: str, mcp_record:` `MCPRecord) -> str`asyncabstractabstractmethod	创建或更新 installed-MCP。 Create or update an installed-MCP record.
`async def list_mcps(user_id: str) -> list[MCPRecord]`asyncabstractabstractmethod	列出 `list_mcps`。 List every MCP a user has installed, enabled or not.
`async def get_mcp(user_id: str, mcp_id: str) ->` `MCPRecord \| None`asyncabstractabstractmethod	获取MCP。 Fetch a single installed-MCP record by id.
`async def get_mcp_by_name(user_id: str, name: str) ->` `MCPRecord \| None`asyncabstractabstractmethod	获取MCP 通过名称。 Fetch an installed-MCP record by its MCP name.
`async def delete_mcp(user_id: str, mcp_id: str) -> bool`asyncabstractabstractmethod	删除 installed-MCP。 Delete an installed-MCP record.
`async def upsert_skill(user_id: str, skill_record:` `SkillRecord) -> str`asyncabstractabstractmethod	执行 `upsert_skill` 方法。 Create or update an installed-skill record.
`async def list_skills(user_id: str) -> list[SkillRecord]`asyncabstractabstractmethod	列出技能。 List every skill a user has installed, enabled or not.
`async def get_skill(user_id: str, skill_id: str) ->` `SkillRecord \| None`asyncabstractabstractmethod	获取技能。 Fetch a single installed-skill record by id.
`async def get_skill_by_name(user_id: str, name: str) ->` `SkillRecord \| None`asyncabstractabstractmethod	获取技能通过名称。 Fetch an installed-skill record by its skill name.
`async def delete_skill(user_id: str, skill_id: str) -> bool`asyncabstractabstractmethod	删除技能。 Delete an installed-skill record.
`async def upsert_agent(user_id: str, agent_record:` `AgentRecord) -> str`asyncabstractabstractmethod	执行 `upsert_agent` 方法。 Create an agent record in the storage.
`async def list_agents(user_id: str) -> list[AgentRecord]`asyncabstractabstractmethod	列出智能体。 List all agents for a given user.
`async def get_agent(user_id: str, agent_id: str) ->` `AgentRecord \| None`asyncabstractabstractmethod	获取智能体。 Fetch a single agent record by id.
`async def delete_agent(user_id: str, agent_id: str) -> bool`asyncabstractabstractmethod	删除智能体。 Delete an agent record.
`async def upsert_session(user_id: str, agent_id: str, config:` `SessionConfig, state: AgentState \| None = None, session_i…`asyncabstractabstractmethod	执行 `upsert_session` 方法。 Create or update a session for a (user, agent) pair.
`async def set_session_team_id(user_id: str, session_id: str, team_id: str \| None) -> None`asyncabstractabstractmethod	设置会话团队标识。 Set or clear ``team_id`` on an existing session record.
`async def update_session_state(user_id: str, agent_id: str, session_id: str, state:` `AgentState) -> None`asyncabstractabstractmethod	更新会话状态。 Update only the mutable state of an existing session.
`async def list_sessions(user_id: str, agent_id: str) -> list[SessionRecord]`asyncabstractabstractmethod	列出会话。 List all sessions for a given user and agent entity.
`async def delete_session(user_id: str, agent_id: str, session_id: str) -> bool`asyncabstractabstractmethod	删除会话。 Delete a session.
`async def get_session(user_id: str, agent_id: str, session_id: str) ->` `SessionRecord \| None`asyncabstractabstractmethod	获取会话。 Fetch a single session record by id.
`async def list_sessions_by_schedule(user_id: str, schedule_id: str) -> list[SessionRecord]`asyncabstractabstractmethod	列出会话通过调度。 Return all sessions created by a given schedule.
`async def list_sessions_by_channel(user_id: str, channel_id: str) -> list[SessionRecord]`asyncabstractabstractmethod	列出会话通过渠道。 Return all sessions derived from a given channel.
`async def upsert_schedule(user_id: str, record:` `ScheduleRecord) -> str`asyncabstractabstractmethod	执行 `upsert_schedule` 方法。 Persist a cron task record and register it in the user's index.
`async def get_schedule(user_id: str, schedule_id: str) ->` `ScheduleRecord \| None`asyncabstractabstractmethod	获取调度。 Fetch a single cron task record by id.
`async def list_schedules(user_id: str) -> list[ScheduleRecord]`asyncabstractabstractmethod	列出调度。 Return all cron task records belonging to the given user.
`async def delete_schedule(user_id: str, schedule_id: str) -> bool`asyncabstractabstractmethod	删除调度。 Delete a cron task record and remove it from the user's index.
`async def list_all_schedules() -> list[ScheduleRecord]`asyncabstractabstractmethod	列出所有调度。 Return every schedule record across all users.
`async def upsert_channel(record:` `ChannelRecord, platform_bot_id: str) -> str`async	执行 `upsert_channel` 方法。 Persist a channel record and refresh its indexes.
`async def get_channel(channel_id: str) ->` `ChannelRecord \| None`async	获取渠道。 Fetch a channel record by its global id.
`async def list_channels(user_id: str) -> list[ChannelRecord]`async	列出渠道。 Return all channel records owned by the given user.
`async def list_all_channels() -> list[ChannelRecord]`async	列出所有渠道。 Return every channel record across all users.
`async def delete_channel(channel_id: str, platform_bot_id: str) -> bool`async	删除渠道。 Delete a channel record and remove it from all indexes.
`async def get_channel_id_by_platform_bot_id(platform_bot_id: str) -> str \| None`async	获取 `get_channel_id_by_platform_bot_id`。 Return the channel id currently bound to a platform bot, if any.
`async def upsert_message(user_id: str, session_id: str, msg:` `Msg) -> None`asyncabstractabstractmethod	执行 `upsert_message` 方法。 Persist a message to the session's message list.
`async def delete_message(user_id: str, session_id: str, message_id: str) -> bool`asyncabstractabstractmethod	删除消息。 Delete one message from a session.
`async def get_message(user_id: str, session_id: str, message_id: str) ->` `Msg \| None`asyncabstractabstractmethod	获取消息。 Fetch a single message by id from the session's message list.
`async def list_messages(user_id: str, session_id: str, limit: int = 50, before: str \| None = None, **kwargs: Any) -> tu…`asyncabstractabstractmethod	列出消息。 Return the most recent messages for a session with cursor-based pagination.
`async def upsert_sop(user_id: str, record:` `SOPRecord) -> SOPRecord`async	执行 `upsert_sop` 方法。 Create or overwrite a procedure.
`async def get_sop(user_id: str, sop_id: str) ->` `SOPRecord \| None`async	获取SOP。 Fetch one procedure; owner-scoped.
`async def list_sops(user_id: str) -> list[SOPRecord]`async	列出 `list_sops`。 List the user's procedures.
`async def delete_sop(user_id: str, sop_id: str) -> bool`async	删除SOP。 Delete a procedure, every run of it, and their conversations.
`async def upsert_sop_run(user_id: str, record:` `SOPRunRecord) -> SOPRunRecord`async	执行 `upsert_sop_run` 方法。 Create or overwrite a run.
`async def get_sop_run(user_id: str, sop_run_id: str) ->` `SOPRunRecord \| None`async	获取SOP 运行。 Fetch one run; owner-scoped.
`async def list_sop_runs(user_id: str, sop_id: str \| None = None, phase:` `SOPPhase \| None = None) -> list[SOPRunRecord]`async	列出 `list_sop_runs`。 List the user's runs, newest first.
`async def update_sop_run(user_id: str, sop_run_id: str, state:` `SOPRunState, sessions: dict[str, str] \| None = None) ->…`async	更新SOP 运行。 Update a run's state (and sessions), leaving its definition.
`async def delete_sop_run(user_id: str, sop_run_id: str) -> bool`async	删除SOP 运行。 Delete one run and the conversations it opened.
`async def upsert_team(user_id: str, record:` `TeamRecord) -> TeamRecord`asyncabstractabstractmethod	创建或更新团队。 Create or update a team record.
`async def get_team(user_id: str, team_id: str) ->` `TeamRecord \| None`asyncabstractabstractmethod	获取团队。 Fetch a single team record by id.
`async def list_teams(user_id: str) -> list[TeamRecord]`asyncabstractabstractmethod	列出 `list_teams`。 List all teams owned by a given user.
`async def delete_team(user_id: str, team_id: str) -> bool`asyncabstractabstractmethod	删除团队。 Delete a team record and cascade-clean its members by role.
`async def upsert_knowledge_base(user_id: str, record:` `KnowledgeBaseRecord) -> KnowledgeBaseRecord`asyncabstractabstractmethod	创建或更新知识库基类。 Create or update a knowledge base record.
`async def get_knowledge_base(user_id: str, knowledge_base_id: str) ->` `KnowledgeBaseRecord \| None`asyncabstractabstractmethod	获取知识库。 Fetch a single knowledge base record by id.
`async def list_knowledge_bases(user_id: str) -> list[KnowledgeBaseRecord]`asyncabstractabstractmethod	列出 `list_knowledge_bases`。 List all knowledge base records owned by the given user.
`async def delete_knowledge_base(user_id: str, knowledge_base_id: str) -> bool`asyncabstractabstractmethod	删除知识库。 Delete a knowledge base record and remove it from the user index.
`async def upsert_knowledge_document(user_id: str, record:` `KnowledgeDocumentRecord) -> KnowledgeDocumentRecord`asyncabstractabstractmethod	创建或更新知识库文档。 Create or update a knowledge document record.
`async def get_knowledge_document(user_id: str, knowledge_base_id: str, document_id: str) ->` `KnowledgeDocumentRecord \| N…`asyncabstractabstractmethod	获取知识库文档。 Fetch a single knowledge document record by id.
`async def list_knowledge_documents(user_id: str, knowledge_base_id: str) -> list[KnowledgeDocumentRecord]`asyncabstractabstractmethod	列出知识库文档。 List all documents in a knowledge base.
`async def delete_knowledge_document(user_id: str, knowledge_base_id: str, document_id: str) -> bool`asyncabstractabstractmethod	删除知识库文档。 Delete a knowledge document record.
`async def update_knowledge_document_status(user_id: str, knowledge_base_id: str, document_id: str, status: KnowledgeDoc…`asyncabstractabstractmethod	更新知识库文档状态。 Update only the status-related fields of a document record.
`async def acquire_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str…`asyncabstractabstractmethod	执行 `acquire_knowledge_document_lease` 方法。 Compare-and-swap acquisition of the document processing lease.
`async def renew_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str,…`asyncabstractabstractmethod	执行 `renew_knowledge_document_lease` 方法。 Extend an existing lease this worker already holds.
`async def release_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str…`asyncabstractabstractmethod	执行 `release_knowledge_document_lease` 方法。 Release the processing lease this worker holds.
`async def list_knowledge_documents_with_expired_lease(now: datetime \| None = None) -> list[KnowledgeDocumentRecord]`asyncabstractabstractmethod	列出 `list_knowledge_documents_with_expired_lease`。 Return non-terminal documents whose lease has expired.
`async def list_knowledge_documents_pending_since(threshold: datetime) -> list[KnowledgeDocumentRecord]`asyncabstractabstractmethod	列出 `list_knowledge_documents_pending_since`。 Return documents stuck in ``pending`` older than ``threshold``.

方法详情Method Detail

aenter

async def __aenter__() -> Self

async

进入异步上下文并返回自身。

Start the storage backend (open connection pool, etc.).

返回Returns Self

返回自身。Returns self.

源码Source agentscope/app/storage/_base.py:35–37

aexit

async def __aexit__(exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: Any) -> None

async

退出异步上下文并清理资源。

Shut down the storage backend.

参数Parameters

名称Name	类型Type	说明Description
`exc_type`	type[BaseException] \| None	exc 类型（类型 `type[BaseException] \| None`）。The exc type.
`exc_value`	BaseException \| None	exc 值（类型 `BaseException \| None`）。The exc value.
`traceback`	Any	traceback（类型 `Any`）。The traceback.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_base.py:39–46

aclose

async def aclose() -> None

async

返回 aclose 结果。

Release underlying connection resources.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_base.py:48–49

upsert_credential

async def upsert_credential(user_id: str, credential_data: CredentialBase) -> str

asyncabstractabstractmethod

创建或更新凭证 in 存储。

Create or update a credential in the storage.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.
`credential_data`	CredentialBase	凭证数据。The credential data.

返回Returns str

返回字符串结果。`str`:

源码Source agentscope/app/storage/_base.py:52–68

list_credentials

async def list_credentials(user_id: str) -> list[CredentialRecord]

asyncabstractabstractmethod

列出凭证。

List all credentials for a given user.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.

返回Returns list[CredentialRecord]

返回CredentialRecord列表。`list[CredentialRecord]`:

源码Source agentscope/app/storage/_base.py:71–81

get_credential

async def get_credential(user_id: str, credential_id: str) -> CredentialRecord | None

asyncabstractabstractmethod

获取凭证。

Fetch a single credential record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`credential_id`	str	凭证标识。The credential id.

返回Returns CredentialRecord | None

返回CredentialRecord|None。`CredentialRecord | None`: The record, or `None` if not found.

源码Source agentscope/app/storage/_base.py:84–97

delete_credential

async def delete_credential(user_id: str, credential_id: str) -> bool

asyncabstractabstractmethod

删除凭证。

Delete a credential.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.
`credential_id`	str	凭证标识。The credential id.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/storage/_base.py:100–116

upsert_mcp

async def upsert_mcp(user_id: str, mcp_record: MCPRecord) -> str

asyncabstractabstractmethod

创建或更新 installed-MCP。

The record's client.name is unique per user — the workspace relation is derived by joining on it — so writing a name another record already holds is an error rather than an overwrite.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.
`mcp_record`	MCPRecord	MCP 记录（类型 `MCPRecord`）。The record to write. Its `id` decides create vs update.

返回Returns str

返回字符串结果。`str`:

异常Raises

可能抛出 `Exception`。`ValueError`:
可能抛出 `Exception`。When another record of this user already uses the name.

源码Source agentscope/app/storage/_base.py:119–139

list_mcps

async def list_mcps(user_id: str) -> list[MCPRecord]

asyncabstractabstractmethod

列出 list_mcps。

List every MCP a user has installed, enabled or not.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.

返回Returns list[MCPRecord]

返回MCPRecord列表。`list[MCPRecord]`:

源码Source agentscope/app/storage/_base.py:142–152

get_mcp

async def get_mcp(user_id: str, mcp_id: str) -> MCPRecord | None

asyncabstractabstractmethod

获取MCP。

Fetch a single installed-MCP record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`mcp_id`	str	MCP 标识（类型 `str`）。The record id.

返回Returns MCPRecord | None

返回MCPRecord|None。`MCPRecord | None`:

源码Source agentscope/app/storage/_base.py:155–167

get_mcp_by_name

async def get_mcp_by_name(user_id: str, name: str) -> MCPRecord | None

asyncabstractabstractmethod

获取MCP 通过名称。

This is the lookup the workspace relation is derived through — a workspace holds names, not record ids.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`name`	str	名称（类型 `str`）。The MCP name, i.e. `record.client.name`.

返回Returns MCPRecord | None

返回MCPRecord|None。`MCPRecord | None`:

源码Source agentscope/app/storage/_base.py:170–189

delete_mcp

async def delete_mcp(user_id: str, mcp_id: str) -> bool

asyncabstractabstractmethod

删除 installed-MCP。

Delete an installed-MCP record.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.
`mcp_id`	str	MCP 标识（类型 `str`）。The record id.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/storage/_base.py:192–204

upsert_skill

async def upsert_skill(user_id: str, skill_record: SkillRecord) -> str

asyncabstractabstractmethod

执行 upsert_skill 方法。

The record's name is unique per user, for the same reason MCP names are: a workspace refers to a skill by name.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.
`skill_record`	SkillRecord	技能记录（类型 `SkillRecord`）。The record to write. Its `id` decides create vs update.

返回Returns str

返回字符串结果。`str`:

异常Raises

可能抛出 `Exception`。`ValueError`:
可能抛出 `Exception`。When another record of this user already uses the name.

源码Source agentscope/app/storage/_base.py:207–230

list_skills

async def list_skills(user_id: str) -> list[SkillRecord]

asyncabstractabstractmethod

列出技能。

Named apart from the workspace's list_skills because these are library records, not skills present in any workspace.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.

返回Returns list[SkillRecord]

返回SkillRecord列表。`list[SkillRecord]`:

源码Source agentscope/app/storage/_base.py:233–246

get_skill

async def get_skill(user_id: str, skill_id: str) -> SkillRecord | None

asyncabstractabstractmethod

获取技能。

Fetch a single installed-skill record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`skill_id`	str	技能标识（类型 `str`）。The record id.

返回Returns SkillRecord | None

返回SkillRecord|None。`SkillRecord | None`:

源码Source agentscope/app/storage/_base.py:249–265

get_skill_by_name

async def get_skill_by_name(user_id: str, name: str) -> SkillRecord | None

asyncabstractabstractmethod

获取技能通过名称。

Fetch an installed-skill record by its skill name.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`name`	str	技能名称。The skill name.

返回Returns SkillRecord | None

返回SkillRecord|None。`SkillRecord | None`:

源码Source agentscope/app/storage/_base.py:268–284

delete_skill

async def delete_skill(user_id: str, skill_id: str) -> bool

asyncabstractabstractmethod

删除技能。

Delete an installed-skill record.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.
`skill_id`	str	技能标识（类型 `str`）。The record id.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/storage/_base.py:287–303

upsert_agent

async def upsert_agent(user_id: str, agent_record: AgentRecord) -> str

asyncabstractabstractmethod

执行 upsert_agent 方法。

Create an agent record in the storage.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.
`agent_record`	AgentRecord	智能体类。The agent record.

返回Returns str

返回字符串结果。`str`:

源码Source agentscope/app/storage/_base.py:306–322

list_agents

async def list_agents(user_id: str) -> list[AgentRecord]

asyncabstractabstractmethod

列出智能体。

List all agents for a given user.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.

返回Returns list[AgentRecord]

返回AgentRecord列表。`list[AgentRecord]`:

源码Source agentscope/app/storage/_base.py:325–335

get_agent

async def get_agent(user_id: str, agent_id: str) -> AgentRecord | None

asyncabstractabstractmethod

获取智能体。

Fetch a single agent record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`agent_id`	str	智能体标识。The agent id.

返回Returns AgentRecord | None

返回AgentRecord|None。`AgentRecord | None`: The record, or `None` if not found.

源码Source agentscope/app/storage/_base.py:338–351

delete_agent

async def delete_agent(user_id: str, agent_id: str) -> bool

asyncabstractabstractmethod

删除智能体。

Delete an agent record.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.
`agent_id`	str	智能体标识。The agent id.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/storage/_base.py:354–366

upsert_session

async def upsert_session(user_id: str, agent_id: str, config: SessionConfig, state: AgentState | None = None, session_id: str | None = None, origin: SessionOrigin | None = None, source: str | None = None, source_schedule_id: str | None = None, source_chat_id: str | None = None, source_chat_name: str | None = None, source_channel_id: str | None = None) -> SessionRecord

asyncabstractabstractmethod

执行 upsert_session 方法。

Create or update a session for a (user, agent) pair.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`agent_id`	str	智能体标识。The agent id.
`config`	SessionConfig	配置对象（类型 `SessionConfig`）。Immutable session configuration (model, workspace). Required on create; passed unchanged on state-only updates.
`state`默认default `None`	AgentState \| None	状态对象（类型 `AgentState \| None`）。Runtime state to persist. Defaults to a fresh `AgentState()` when `None`.
`session_id`默认default `None`	str \| None	会话标识（类型 `str \| None`）。If provided, update the existing session with this id. If `None`, create a new session.
`origin`默认default `None`	SessionOrigin \| None	来源（类型 `SessionOrigin \| None`）。How the session came to exist — a :class:`ScheduleOrigin` also indexes it under its schedule. Defaults to :class:`UserOrigin`. source / source_schedule_id / source_chat_id / source_chat_name / source_channel_id: Deprecated — the flat shape `origin` replaced. Passing any of them still builds the matching origin, so callers written against the old signature keep working.
`source`默认default `None`	str \| None	来源（类型 `str \| None`）。The source.
`source_schedule_id`默认default `None`	str \| None	来源调度标识（类型 `str \| None`）。The source schedule id.
`source_chat_id`默认default `None`	str \| None	来源聊天标识（类型 `str \| None`）。The source chat id.
`source_chat_name`默认default `None`	str \| None	来源聊天名称（类型 `str \| None`）。The source chat name.
`source_channel_id`默认default `None`	str \| None	来源渠道标识（类型 `str \| None`）。The source channel id.

返回Returns SessionRecord

`SessionRecord`:已创建或已更新。`SessionRecord`: The created or updated record.

源码Source agentscope/app/storage/_base.py:369–407

set_session_team_id

async def set_session_team_id(user_id: str, session_id: str, team_id: str | None) -> None

asyncabstractabstractmethod

设置会话团队标识。

Bypasses :meth:upsert_session because that method does not write team_id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`session_id`	str	会话标识（类型 `str`）。The session whose `team_id` should be updated.
`team_id`	str \| None	团队标识（类型 `str \| None`）。The new value. `None` detaches the session from any team.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_base.py:410–432

update_session_state

async def update_session_state(user_id: str, agent_id: str, session_id: str, state: AgentState) -> None

asyncabstractabstractmethod

更新会话状态。

Convenience method for the hot path (post-chat-turn persistence).

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`agent_id`	str	智能体标识。The agent id.
`session_id`	str	会话标识。The session id.
`state`	AgentState	状态对象（类型 `AgentState`）。The new agent state to persist.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_base.py:435–452

list_sessions

async def list_sessions(user_id: str, agent_id: str) -> list[SessionRecord]

asyncabstractabstractmethod

列出会话。

List all sessions for a given user and agent entity.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.
`agent_id`	str	智能体标识。The agent id.

返回Returns list[SessionRecord]

用于(用户,智能体)的`list[SessionRecord]`:列出的所有会话。`list[SessionRecord]`: List of all sessions for the (user, agent).

源码Source agentscope/app/storage/_base.py:455–468

delete_session

async def delete_session(user_id: str, agent_id: str, session_id: str) -> bool

asyncabstractabstractmethod

删除会话。

Delete a session.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.
`agent_id`	str	智能体标识。The agent id.
`session_id`	str	会话标识。The session id.

返回Returns bool

返回布尔值，表示判断结果。`bool`: True if deleted, False if not found.

源码Source agentscope/app/storage/_base.py:471–486

get_session

async def get_session(user_id: str, agent_id: str, session_id: str) -> SessionRecord | None

asyncabstractabstractmethod

获取会话。

Fetch a single session record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`agent_id`	str	智能体标识。The agent id.
`session_id`	str	会话标识。The session id.

返回Returns SessionRecord | None

返回SessionRecord|None。`SessionRecord | None`: The record, or `None` if not found.

源码Source agentscope/app/storage/_base.py:489–504

list_sessions_by_schedule

async def list_sessions_by_schedule(user_id: str, schedule_id: str) -> list[SessionRecord]

asyncabstractabstractmethod

列出会话通过调度。

Return all sessions created by a given schedule.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`schedule_id`	str	调度标识。The schedule id.

返回Returns list[SessionRecord]

返回SessionRecord列表。`list[SessionRecord]`: Sessions triggered by this schedule,

源码Source agentscope/app/storage/_base.py:507–521

list_sessions_by_channel

async def list_sessions_by_channel(user_id: str, channel_id: str) -> list[SessionRecord]

asyncabstractabstractmethod

列出会话通过渠道。

Return all sessions derived from a given channel.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`channel_id`	str	渠道标识。The channel id.

返回Returns list[SessionRecord]

返回SessionRecord列表。`list[SessionRecord]`: Sessions the channel spawned, ordered

源码Source agentscope/app/storage/_base.py:524–538

upsert_schedule

async def upsert_schedule(user_id: str, record: ScheduleRecord) -> str

asyncabstractabstractmethod

执行 upsert_schedule 方法。

Persist a cron task record and register it in the user's index.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`record`	ScheduleRecord	记录（类型 `ScheduleRecord`）。The fully-populated record to store.

返回Returns str

`str`:标识的已存储。`str`: The id of the stored record.

源码Source agentscope/app/storage/_base.py:541–554

get_schedule

async def get_schedule(user_id: str, schedule_id: str) -> ScheduleRecord | None

asyncabstractabstractmethod

获取调度。

Fetch a single cron task record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`schedule_id`	str	任务标识。The task id.

返回Returns ScheduleRecord | None

返回ScheduleRecord|None。`ScheduleRecord | None`: The record, or `None` if not found.

源码Source agentscope/app/storage/_base.py:557–570

list_schedules

async def list_schedules(user_id: str) -> list[ScheduleRecord]

asyncabstractabstractmethod

列出调度。

Return all cron task records belonging to the given user.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.

返回Returns list[ScheduleRecord]

返回ScheduleRecord列表。`list[ScheduleRecord]`: All cron task records for the user.

源码Source agentscope/app/storage/_base.py:573–584

delete_schedule

async def delete_schedule(user_id: str, schedule_id: str) -> bool

asyncabstractabstractmethod

删除调度。

Delete a cron task record and remove it from the user's index.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`schedule_id`	str	任务到删除的标识。The id of the task to delete.

返回Returns bool

返回布尔值，表示判断结果。`bool`: `True` if deleted, `False` if not found.

源码Source agentscope/app/storage/_base.py:587–600

list_all_schedules

async def list_all_schedules() -> list[ScheduleRecord]

asyncabstractabstractmethod

列出所有调度。

Used on startup to restore the in-memory scheduler from persisted state.

返回Returns list[ScheduleRecord]

返回ScheduleRecord列表。`list[ScheduleRecord]`: All schedule records in the store.

源码Source agentscope/app/storage/_base.py:603–611

upsert_channel

async def upsert_channel(record: ChannelRecord, platform_bot_id: str) -> str

async

执行 upsert_channel 方法。

record.id is a globally unique UUID, so the record lives at a single global key; record.user_id drives the per-user index and platform_bot_id (extracted from credentials by the caller) drives the uniqueness index.

参数Parameters

名称Name	类型Type	说明Description
`record`	ChannelRecord	记录（类型 `ChannelRecord`）。The channel record to store.
`platform_bot_id`	str	平台 bot 标识（类型 `str`）。The platform-side bot identifier, used to maintain the dedup index.

返回Returns str

`str`:标识的已存储。`str`: The id of the stored record.

源码Source agentscope/app/storage/_base.py:623–645

get_channel

async def get_channel(channel_id: str) -> ChannelRecord | None

async

获取渠道。

This is the primary lookup, used both by the management API and by the channel runtime (which only has a channel_id in hand).

参数Parameters

名称Name	类型Type	说明Description
`channel_id`	str	渠道标识。The channel id.

返回Returns ChannelRecord | None

返回ChannelRecord|None。`ChannelRecord | None`: The record, or `None` if not found.

源码Source agentscope/app/storage/_base.py:647–664

list_channels

async def list_channels(user_id: str) -> list[ChannelRecord]

async

列出渠道。

Return all channel records owned by the given user.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.

返回Returns list[ChannelRecord]

返回ChannelRecord列表。`list[ChannelRecord]`: All channel records for the user.

源码Source agentscope/app/storage/_base.py:666–680

list_all_channels

async def list_all_channels() -> list[ChannelRecord]

async

列出所有渠道。

Used on startup / reconcile to restore channel instances.

返回Returns list[ChannelRecord]

返回ChannelRecord列表。`list[ChannelRecord]`: All channel records in the store.

源码Source agentscope/app/storage/_base.py:682–692

delete_channel

async def delete_channel(channel_id: str, platform_bot_id: str) -> bool

async

删除渠道。

Delete a channel record and remove it from all indexes.

参数Parameters

名称Name	类型Type	说明Description
`channel_id`	str	渠道到删除的标识。The id of the channel to delete.
`platform_bot_id`	str	平台 bot 标识（类型 `str`）。The bot identifier (re-extracted from credentials by the caller) so the dedup index entry can be removed.

返回Returns bool

返回布尔值，表示判断结果。`bool`: `True` if deleted, `False` if not found.

源码Source agentscope/app/storage/_base.py:694–712

get_channel_id_by_platform_bot_id

async def get_channel_id_by_platform_bot_id(platform_bot_id: str) -> str | None

async

获取 get_channel_id_by_platform_bot_id。

Used for uniqueness validation — no two channels may share the same platform_bot_id.

参数Parameters

名称Name	类型Type	说明Description
`platform_bot_id`	str	平台 bot 标识（类型 `str`）。The platform-side bot identifier.

返回Returns str | None

返回str|None。`str | None`: The bound channel id, or `None`.

源码Source agentscope/app/storage/_base.py:714–731

upsert_message

async def upsert_message(user_id: str, session_id: str, msg: Msg) -> None

asyncabstractabstractmethod

执行 upsert_message 方法。

If a message in the list has the same id as msg, it is replaced in place.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`session_id`	str	会话标识。The session id.
`msg`	Msg	消息（类型 `Msg`）。The message to persist.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_base.py:738–753

delete_message

async def delete_message(user_id: str, session_id: str, message_id: str) -> bool

asyncabstractabstractmethod

删除消息。

Delete one message from a session.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`session_id`	str	会话标识。The session id.
`message_id`	str	消息标识到删除。The message id to delete.

返回Returns bool

返回布尔值，表示判断结果。`bool`: Whether a matching message was deleted.

源码Source agentscope/app/storage/_base.py:756–771

get_message

async def get_message(user_id: str, session_id: str, message_id: str) -> Msg | None

asyncabstractabstractmethod

获取消息。

Fetch a single message by id from the session's message list.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`session_id`	str	会话标识。The session id.
`message_id`	str	消息标识（类型 `str`）。The message id to look up.

返回Returns Msg | None

返回消息|None。`Msg | None`: The message, or `None` if not found.

源码Source agentscope/app/storage/_base.py:774–789

list_messages

async def list_messages(user_id: str, session_id: str, limit: int = 50, before: str | None = None, **kwargs: Any) -> tuple[list[Msg], bool]

asyncabstractabstractmethod

列出消息。

Return the most recent messages for a session with cursor-based pagination.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`session_id`	str	会话标识。The session id.
`limit`默认default `50`	int	数量上限（类型 `int`）。Maximum number of messages to return. Defaults to 50.
`before`默认default `None`	str \| None	之前（类型 `str \| None`）。A message ID used as the cursor. When provided, returns messages created before this message. Omit to get the latest page.
`kwargs`	Any	关键字参数（类型 `Any`）。The kwargs.

返回Returns tuple[list[Msg], bool]

(消息 in的`tuple[list[Msg], bool]`:元组。`tuple[list[Msg], bool]`: A tuple of (messages in

源码Source agentscope/app/storage/_base.py:792–819

upsert_sop

async def upsert_sop(user_id: str, record: SOPRecord) -> SOPRecord

async

执行 upsert_sop 方法。

Create or overwrite a procedure.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`record`	SOPRecord	记录（类型 `SOPRecord`）。The procedure to store.

返回Returns SOPRecord

返回SOPRecord。`SOPRecord`:

源码Source agentscope/app/storage/_base.py:825–838

get_sop

async def get_sop(user_id: str, sop_id: str) -> SOPRecord | None

async

获取SOP。

Fetch one procedure; owner-scoped.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`sop_id`	str	SOP 标识（类型 `str`）。The procedure id.

返回Returns SOPRecord | None

返回SOPRecord|None。`SOPRecord | None`:

源码Source agentscope/app/storage/_base.py:840–853

list_sops

async def list_sops(user_id: str) -> list[SOPRecord]

async

列出 list_sops。

List the user's procedures.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.

返回Returns list[SOPRecord]

返回SOPRecord列表。`list[SOPRecord]`:

源码Source agentscope/app/storage/_base.py:855–866

delete_sop

async def delete_sop(user_id: str, sop_id: str) -> bool

async

删除SOP。

Delete a procedure, every run of it, and their conversations.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`sop_id`	str	SOP 标识（类型 `str`）。The procedure id.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/storage/_base.py:868–881

upsert_sop_run

async def upsert_sop_run(user_id: str, record: SOPRunRecord) -> SOPRunRecord

async

执行 upsert_sop_run 方法。

Create or overwrite a run.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`record`	SOPRunRecord	运行到存储。The run to store.

返回Returns SOPRunRecord

返回SOPRunRecord。`SOPRunRecord`:

源码Source agentscope/app/storage/_base.py:883–900

get_sop_run

async def get_sop_run(user_id: str, sop_run_id: str) -> SOPRunRecord | None

async

获取SOP 运行。

Fetch one run; owner-scoped.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`sop_run_id`	str	运行标识。The run id.

返回Returns SOPRunRecord | None

返回SOPRunRecord|None。`SOPRunRecord | None`:

源码Source agentscope/app/storage/_base.py:902–919

list_sop_runs

async def list_sop_runs(user_id: str, sop_id: str | None = None, phase: SOPPhase | None = None) -> list[SOPRunRecord]

async

列出 list_sop_runs。

List the user's runs, newest first.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`sop_id`默认default `None`	str \| None	SOP 标识（类型 `str \| None`）。Only runs of this procedure. `None` means all of them.
`phase`默认default `None`	SOPPhase \| None	阶段（类型 `SOPPhase \| None`）。Only runs in this phase. `None` means all of them.

返回Returns list[SOPRunRecord]

返回SOPRunRecord列表。`list[SOPRunRecord]`:

源码Source agentscope/app/storage/_base.py:921–941

update_sop_run

async def update_sop_run(user_id: str, sop_run_id: str, state: SOPRunState, sessions: dict[str, str] | None = None) -> None

async

更新SOP 运行。

Update a run's state (and sessions), leaving its definition.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`sop_run_id`	str	运行标识。The run id.
`state`	SOPRunState	状态对象（类型 `SOPRunState`）。How the run is going now.
`sessions`默认default `None`	dict[str, str] \| None	会话（类型 `dict[str, str] \| None`）。The run's conversations. `None` leaves them as is.

返回Returns None

无返回值。No value is returned.

异常Raises

可能抛出 `Exception`。`KeyError`:
可能抛出 `Exception`。If the user has no such run.

源码Source agentscope/app/storage/_base.py:943–966

delete_sop_run

async def delete_sop_run(user_id: str, sop_run_id: str) -> bool

async

删除SOP 运行。

Delete one run and the conversations it opened.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`sop_run_id`	str	运行标识。The run id.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/storage/_base.py:968–981

upsert_team

async def upsert_team(user_id: str, record: TeamRecord) -> TeamRecord

asyncabstractabstractmethod

创建或更新团队。

Create or update a team record.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`record`	TeamRecord	记录（类型 `TeamRecord`）。The team record to persist. The record's `id` is used as the primary key; if a record with the same id already exists it is overwritten.

返回Returns TeamRecord

返回TeamRecord。`TeamRecord`: The stored record (with `updated_at` refreshed).

源码Source agentscope/app/storage/_base.py:988–1003

get_team

async def get_team(user_id: str, team_id: str) -> TeamRecord | None

asyncabstractabstractmethod

获取团队。

Fetch a single team record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`team_id`	str	团队标识。The team id.

返回Returns TeamRecord | None

返回TeamRecord|None。`TeamRecord | None`: The record, or `None` if not found.

源码Source agentscope/app/storage/_base.py:1006–1019

list_teams

async def list_teams(user_id: str) -> list[TeamRecord]

asyncabstractabstractmethod

列出 list_teams。

List all teams owned by a given user.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识。The user id.

返回Returns list[TeamRecord]

返回TeamRecord列表。`list[TeamRecord]`: All team records belonging to the user.

源码Source agentscope/app/storage/_base.py:1022–1030

delete_team

async def delete_team(user_id: str, team_id: str) -> bool

asyncabstractabstractmethod

删除团队。

The cascade is role-aware — the two team-membership modes (created vs invited, see :class:TeamMember) must be handled differently: 1.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`team_id`	str	团队到删除的标识。The id of the team to delete.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/storage/_base.py:1033–1070

upsert_knowledge_base

async def upsert_knowledge_base(user_id: str, record: KnowledgeBaseRecord) -> KnowledgeBaseRecord

asyncabstractabstractmethod

创建或更新知识库基类。

The caller is responsible for constructing the full :class:KnowledgeBaseRecord (including id and collection_name).

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The owner user id. Must match `record.user_id`.
`record`	KnowledgeBaseRecord	记录（类型 `KnowledgeBaseRecord`）。The fully-populated record to persist.

返回Returns KnowledgeBaseRecord

返回KnowledgeBaseRecord。`KnowledgeBaseRecord`:

源码Source agentscope/app/storage/_base.py:1077–1098

get_knowledge_base

async def get_knowledge_base(user_id: str, knowledge_base_id: str) -> KnowledgeBaseRecord | None

asyncabstractabstractmethod

获取知识库。

Fetch a single knowledge base record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`knowledge_base_id`	str	知识库基类标识。The knowledge base id.

返回Returns KnowledgeBaseRecord | None

返回KnowledgeBaseRecord|None。`KnowledgeBaseRecord | None`:

源码Source agentscope/app/storage/_base.py:1101–1118

list_knowledge_bases

async def list_knowledge_bases(user_id: str) -> list[KnowledgeBaseRecord]

asyncabstractabstractmethod

列出 list_knowledge_bases。

List all knowledge base records owned by the given user.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.

返回Returns list[KnowledgeBaseRecord]

返回KnowledgeBaseRecord列表。`list[KnowledgeBaseRecord]`:

源码Source agentscope/app/storage/_base.py:1121–1134

delete_knowledge_base

async def delete_knowledge_base(user_id: str, knowledge_base_id: str) -> bool

asyncabstractabstractmethod

删除知识库。

Note: this only removes the metadata record; deletion of the underlying vector store collection is the caller's responsibility (typically the knowledge base manager).

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`knowledge_base_id`	str	知识库标识（类型 `str`）。The id of the record to delete.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/storage/_base.py:1137–1158

upsert_knowledge_document

async def upsert_knowledge_document(user_id: str, record: KnowledgeDocumentRecord) -> KnowledgeDocumentRecord

asyncabstractabstractmethod

创建或更新知识库文档。

Used by the upload endpoint to register a freshly arrived document (status='pending') and by other code paths that need to overwrite the full record.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The owner user id. Must match `record.user_id`.
`record`	KnowledgeDocumentRecord	记录（类型 `KnowledgeDocumentRecord`）。The fully-populated record to persist.

返回Returns KnowledgeDocumentRecord

返回KnowledgeDocumentRecord。`KnowledgeDocumentRecord`:

源码Source agentscope/app/storage/_base.py:1165–1187

get_knowledge_document

async def get_knowledge_document(user_id: str, knowledge_base_id: str, document_id: str) -> KnowledgeDocumentRecord | None

asyncabstractabstractmethod

获取知识库文档。

Fetch a single knowledge document record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`knowledge_base_id`	str	父级知识库基类标识。The parent knowledge base id.
`document_id`	str	文档标识。The document id.

返回Returns KnowledgeDocumentRecord | None

返回KnowledgeDocumentRecord|None。`KnowledgeDocumentRecord | None`:

源码Source agentscope/app/storage/_base.py:1190–1209

list_knowledge_documents

async def list_knowledge_documents(user_id: str, knowledge_base_id: str) -> list[KnowledgeDocumentRecord]

asyncabstractabstractmethod

列出知识库文档。

List all documents in a knowledge base.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`knowledge_base_id`	str	父级知识库基类标识。The parent knowledge base id.

返回Returns list[KnowledgeDocumentRecord]

返回KnowledgeDocumentRecord列表。`list[KnowledgeDocumentRecord]`:

源码Source agentscope/app/storage/_base.py:1212–1229

delete_knowledge_document

async def delete_knowledge_document(user_id: str, knowledge_base_id: str, document_id: str) -> bool

asyncabstractabstractmethod

删除知识库文档。

Only removes the metadata record; cleanup of the underlying blob and vector store records is the caller's responsibility.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`knowledge_base_id`	str	父级知识库基类标识。The parent knowledge base id.
`document_id`	str	文档标识（类型 `str`）。The id of the record to delete.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/storage/_base.py:1232–1255

update_knowledge_document_status

async def update_knowledge_document_status(user_id: str, knowledge_base_id: str, document_id: str, status: KnowledgeDocumentStatus, error: str | None = None, chunk_count: int | None = None) -> None

asyncabstractabstractmethod

更新知识库文档状态。

Used by the indexing worker as it walks the lifecycle transitions (parsing → chunking → indexing → ready / error).

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`knowledge_base_id`	str	父级知识库基类标识。The parent knowledge base id.
`document_id`	str	文档标识（类型 `str`）。The document being updated.
`status`	KnowledgeDocumentStatus	新的生命周期状态。The new lifecycle state.
`error`默认default `None`	str \| None	错误（类型 `str \| None`）。Failure reason, set when `status == 'error'`. Ignored otherwise. `None` leaves the existing value unchanged.
`chunk_count`默认default `None`	int \| None	分块数量（类型 `int \| None`）。The final chunk count, set when `status == 'ready'`. `None` leaves the existing value unchanged.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_base.py:1258–1291

acquire_knowledge_document_lease

async def acquire_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str, lease_ttl: timedelta, now: datetime | None = None) -> bool

asyncabstractabstractmethod

执行 acquire_knowledge_document_lease 方法。

Succeeds only if no other worker currently holds a live lease — i.e.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`knowledge_base_id`	str	父级知识库基类标识。The parent knowledge base id.
`document_id`	str	文档标识（类型 `str`）。The document whose lease is being acquired.
`processing_node`	str	processing 节点（类型 `str`）。Stable identifier of the calling worker (e.g. `hostname:pid:uuid`). Persisted as `processing_node` on success.
`lease_ttl`	timedelta	lease 存活时间（类型 `timedelta`）。How long the lease should live from `now`.
`now`默认default `None`	datetime \| None	当前时间（类型 `datetime \| None`）。Reference time for the comparison. Defaults to `datetime.now()`. Injectable for testing.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/storage/_base.py:1294–1334

renew_knowledge_document_lease

async def renew_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str, lease_ttl: timedelta, now: datetime | None = None) -> bool

asyncabstractabstractmethod

执行 renew_knowledge_document_lease 方法。

Required for long-running parses that exceed lease_ttl — the worker calls this periodically so the sweeper does not mistake it for a crash.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`knowledge_base_id`	str	父级知识库基类标识。The parent knowledge base id.
`document_id`	str	文档标识（类型 `str`）。The document being renewed.
`processing_node`	str	processing 节点（类型 `str`）。The caller's processing node id; must match the record's current `processing_node` for renewal to succeed.
`lease_ttl`	timedelta	lease 存活时间（类型 `timedelta`）。The new lease duration relative to `now`.
`now`默认default `None`	datetime \| None	当前时间（类型 `datetime \| None`）。Reference time. Defaults to `datetime.now()`.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/storage/_base.py:1337–1375

release_knowledge_document_lease

async def release_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str) -> None

asyncabstractabstractmethod

执行 release_knowledge_document_lease 方法。

Clears processing_node and data.lease_expires_at only if the current holder matches processing_node — a stolen lease (e.g.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`knowledge_base_id`	str	父级知识库基类标识。The parent knowledge base id.
`document_id`	str	文档标识（类型 `str`）。The document whose lease is being released.
`processing_node`	str	processing 节点（类型 `str`）。The caller's processing node id; must match the record's current holder.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_base.py:1378–1402

list_knowledge_documents_with_expired_lease

async def list_knowledge_documents_with_expired_lease(now: datetime | None = None) -> list[KnowledgeDocumentRecord]

asyncabstractabstractmethod

列出 list_knowledge_documents_with_expired_lease。

Scans every user / knowledge base — used by the sweeper, not by user-facing endpoints.

参数Parameters

名称Name	类型Type	说明Description
`now`默认default `None`	datetime \| None	当前时间（类型 `datetime \| None`）。Reference time. Defaults to `datetime.now()`.

返回Returns list[KnowledgeDocumentRecord]

返回KnowledgeDocumentRecord列表。`list[KnowledgeDocumentRecord]`:

源码Source agentscope/app/storage/_base.py:1405–1425

list_knowledge_documents_pending_since

async def list_knowledge_documents_pending_since(threshold: datetime) -> list[KnowledgeDocumentRecord]

asyncabstractabstractmethod

列出 list_knowledge_documents_pending_since。

Catches the corner case where the upload endpoint persisted a record but the dispatcher (or the process holding it) died before any worker picked the document up — a crashed lease sweep would miss it because no lease was ever written.

参数Parameters

名称Name	类型Type	说明Description
`threshold`	datetime	阈值（类型 `datetime`）。Cut-off creation time; only `pending` records `created_at < threshold` are returned.

返回Returns list[KnowledgeDocumentRecord]

返回KnowledgeDocumentRecord列表。`list[KnowledgeDocumentRecord]`:

源码Source agentscope/app/storage/_base.py:1428–1447