目标: 保护用户数据，防止未授权访问
所有数据通过 Scope 隔离:
from pydantic import BaseModel
class Scope(BaseModel):
user_id: str # 必填 — 用户隔离
agent_id: str | None = None # 可选 — Agent 隔离
team_id: str | None = None # 可选 — 团队隔离
| 规则 |
描述 |
| user_id 必填 |
所有操作必须提供 user_id |
| Scope 验证 |
每次查询自动附加 scope 过滤 |
| 跨 Scope 禁止 |
不允许访问其他用户/agent 数据 |
-- 所有查询自动附加 WHERE 子句
WHERE user_id = ?
AND (agent_id = ? OR agent_id IS NULL)
AND (team_id = ? OR team_id IS NULL)
所有外部输入必须通过 Pydantic model 验证:
from pydantic import BaseModel, Field
from typing import Optional
class CreateDocumentRequest(BaseModel):
user_id: str = Field(..., min_length=1)
agent_id: Optional[str] = None
team_id: Optional[str] = None
content: str = Field(..., min_length=1)
metadata: Optional[dict] = None
| 层级 |
验证内容 |
| MCP 工具 |
参数类型、必填、枚举值 |
| SQLite CRUD |
ID 格式、Scope 匹配 |
| L1 Index |
Vector search uses SQLite + Ollama |
- ❌ SQL 注入字符 (未转义)
- ❌ 路径遍历 (
../)
- ❌ 超长字符串 (>1MB)
- ❌ 无效 URI 格式
不泄露内部细节:
# 安全
raise ValueError("user_id is required for document. Provide scope: { 'user_id': '...' }")
# 不安全 — 泄露 SQL 细节
raise ValueError(f"SQL failed: SELECT * FROM documents WHERE user_id='{user_id}'")
| 类型 |
返回给用户 |
记录到日志 |
| 验证错误 |
✅ 详细消息 |
warn |
| 业务错误 |
✅ 用户友好消息 |
warn |
| 系统错误 |
❌ 通用消息 |
error + 堆栈 |
| 数据类型 |
原因 |
| API Keys |
泄露风险 |
| Passwords |
应使用哈希 |
| Access Tokens |
有过期机制 |
| PII (个人身份信息) |
隐私合规 |
| 记录内容 |
不记录内容 |
| 操作类型 (create/read/update/delete) |
请求体中的敏感字段 |
| 资源类型和 ID |
API Keys |
| Scope (userId, agentId, teamId) |
密码、Token |
| 时间戳 |
完整错误堆栈 (仅内部) |
- 提交
uv.lock 或 poetry.lock 到 Git
- 使用固定版本，不接受范围
| 依赖 |
用途 |
mcp |
MCP 协议 |
pydantic |
Schema 验证 |
aiosqlite |
SQLite (异步) |
- 运行在 stdio (非 HTTP)
- 仅接受本地进程通信
- 不暴露网络端口
# 数据库文件仅所有者可读写
chmod 600 ~/.agents_mem/agents_mem.db
| 变量 |
用途 |
默认值 |
OLLAMA_HOST |
Ollama 地址 |
http://localhost:11434 |
OLLAMA_HOST |
Ollama 地址 |
http://localhost:11434 |
不要在代码中硬编码密钥