- 后端新增 featured_order 字段及精选排序 API(/agents/selector、/agents/{id}/featured、/agents/featured/order),移除 user_agent_pin 表
- 前端新对话页 (ChatPage) 与输入框 (InputBox) 接入 agent-selector,选中后跳转 AgentChatPage 并通过 sessionStorage 透传首条消息
- 新增管理员智能体管理弹窗 (agent-manage-dialog) 支持精选添加/移除/上下移动
84 lines
3.1 KiB
Python
84 lines
3.1 KiB
Python
"""Abstract persistence API for custom agents."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any
|
|
|
|
|
|
class AgentStore(ABC):
|
|
@abstractmethod
|
|
async def list_visible(self, user_id: str) -> list[dict[str, Any]]:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def get_visible(self, agent_id: str, user_id: str) -> dict[str, Any] | None:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def get_any(self, agent_id: str) -> dict[str, Any] | None:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def get_owned(self, agent_id: str, user_id: str) -> dict[str, Any] | None:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def create(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def ensure_builtin(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def update(self, agent_id: str, user_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def update_builtin(self, agent_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
|
|
"""Update a built-in agent (user_id IS NULL).
|
|
|
|
Caller is responsible for authorization — this method does NOT check
|
|
any user identity and returns ``None`` if the target is not a built-in.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def delete(self, agent_id: str, user_id: str) -> bool:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def delete_builtin(self, agent_id: str) -> bool:
|
|
"""Delete a built-in agent (user_id IS NULL).
|
|
|
|
Caller is responsible for authorization — this method does NOT check
|
|
any user identity and returns ``False`` if the target is not a built-in.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def count_owned(self, user_id: str) -> int:
|
|
"""Return the number of agents owned by ``user_id`` (user_id IS NOT NULL)."""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def set_featured(self, agent_id: str, order: int | None) -> dict[str, Any] | None:
|
|
"""Set or clear the homepage-selector featured rank for an agent.
|
|
|
|
Returns the updated row, or ``None`` if the agent does not exist.
|
|
Caller is responsible for authorization (admin) and for verifying
|
|
that ``published`` is true when ``order`` is not ``None``.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def list_featured(self, user_id: str) -> list[dict[str, Any]]:
|
|
"""Return featured + still-visible-to-``user_id`` agents, ordered by ``featured_order`` asc."""
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def list_by_ids(self, agent_ids: list[str], user_id: str) -> list[dict[str, Any]]:
|
|
"""Return rows for ``agent_ids`` visible to ``user_id``, preserving the input order."""
|
|
raise NotImplementedError
|