ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

AgentConnect:用FastAPI与SQLite实现多Agent权限隔离

AgentConnect:用FastAPI与SQLite实现多Agent权限隔离 多 Agent 应用从 Demo 走向团队协作时最常被低估的问题不是模型能力而是权限。AgentConnect 这个项目名称可以理解为一套常见设计思路多个用户可以共享同一个 Agent 实例但每个用户、每个团队能执行的操作和能接触的数据范围必须相互隔离。这个需求表面上看只是加一个权限字段实际落地却涉及身份识别、权限判定、共享关系、缓存失效和数据范围过滤一整条链路。下面从权限模型开始用 FastAPI 和 SQLite 实现一个最小可运行的 AgentConnect 后端再说明权限判定应该放在哪一层、常见事故怎么排查、生产环境还需要补哪些能力。1. AgentConnect 要解决的是共享与权限同时成立的工程问题1.1 为什么共享 Agent 之后真正难的是权限在多数 Agent 项目早期每个开发者都是在本地跑自己的 AgentPrompt、工具调用、环境变量全部私有。一旦进入团队协作阶段就会自然产生一个需求把已经调通的 Agent 共享给别人用。共享之后问题立刻变成一组权限问题谁能调用这个 Agent谁能修改它的 Prompt 和工具配置谁能查看它的运行日志谁能继续把它分享给第三个人日志里包含的数据范围其他人是否都允许看到如果所有人共用同一个 API Key 或同一个服务账号上面的问题都没有答案。系统能知道发生了调用但说不清是谁调用的、以什么权限调用的、应该允许还是拒绝。这也是“共享 Agent 独立权限”这个组合的真实价值Agent 实例是共享资源调用者的身份和授权边界是独立隔离的。1.2 共享、隔离与可审计三者必须同时成立共享意味着资源是公用的隔离意味着请求之间不能越权可审计意味着每一次决策都要能回答“为什么允许”或“为什么拒绝”。这三者同时成立才能支撑 Agent 在多人环境里稳定运行。以一段普通的协作场景为例Alice 创建了一个账单分析 Agent它内部需要读取订单表、调用外部 API、生成汇总报告。B 团队通过团队级授权拿到了调用权限C 用户只被允许查看运行日志。如果权限设计不到位最简单的实现会把整个 Agent 的能力暴露给所有共享者所有人能调用、能改配置、能看全部日志。这在数据上等于把内部接口直接暴露出去。从工程实现看真正要解决的是两件事谁的请求能对哪个 Agent 做什么操作。请求到达 Agent 内部之后工具和数据访问仍然要按同一套身份和范围约束执行。1.3 从模型到代码本文的演示边界下面用一个 FastAPI SQLite 的演示项目来复现 AgentConnect 的最小链路。项目覆盖用户登录、Agent 创建、用户级和团队级分享、动作级权限判定、调用、配置修改、权限回收。SQLite 用于演示数据存储真实生产可以替换成 PostgreSQL 等数据库。演示不包含真实的大模型调用Agent 的执行接口会返回一个模拟结果方便验证权限链路。登录也使用演示 Token不接 SSO但会说明生产环境应该怎么替换。2. 权限模型先定清楚主体、资源、动作与范围2.1 四元组是权限判断的最小骨架在设计 AgentConnect 的权限模型时建议先定义一组四元组维度含义示例主体 Subject发起请求的人或团队user-alice、team-ops资源 Resource要访问的 Agentbilling-helper动作 Action对 Agent 执行的操作invoke、view_log、manage_config、share范围 Scope操作允许触及的数据范围只允许读取本团队账单不允许导出原始表Action 是权限系统最容易定义的部分Scope 却最容易被漏掉。很多权限事故不是用户没有权限而是有权限但没有范围限制。一个用户能调用“账单分析 Agent”如果 Agent 内部查询 SQL 没有按用户归属过滤他就能通过构造输入看到其他客户的数据。因此生产模型里通常需要在授权记录上附带一条条件表达式例如 JSON 形式的策略{scope: team_id team-ops}。本文演示先聚焦 ActionScope 在后续章节说明落地方式。2.2 三种共享粒度用户级、团队级与公开级共享粒度决定了授权记录的维护成本和控制力度。共享粒度使用场景权限维护风险控制用户级一对一协作者每增加一个用户就加一条授权精准但用户多时维护成本高团队级按组织或项目组共享成员关系自动继承团队授权成员变动自动生效但团队范围必须准确公开级所有登录用户都可以发现和使用不需要逐个授权只需要默认策略必须配合默认拒绝和严格 Scope风险最高公开级推荐的做法是没有显式授权就不允许任何操作。也就是说即使 Agent 设置为“公开”也只代表它可以被发现不代表调用者拥有全部动作。真正的动作权限仍然要走统一的权限判定。2.3 所有者、显式授权与团队授权如何合并权限判定需要一个明确的合并规则。本文采用正向授权模型规则如下所有者拥有全部权限包括删除 Agent。显式用户授权优先被读取。团队成员授权通过用户-团队关系推导。合并结果取并集没有任何授权则默认拒绝。一旦引入“拒绝”规则判定顺序应为拒绝优先于允许。动作之间还存在隐含关系。例如允许share应该隐含允许manage_config、view_log、invoke允许manage_config应该隐含view_log和invoke。用一个动作映射表可以避免每个接口都写一遍复杂的权限判断。被授予动作实际可用动作invokeinvokeview_logview_log、invokemanage_configmanage_config、view_log、invokeshareshare、manage_config、view_log、invoke这样 Alice 给 Bob 授view_log权限Bob 可以读取日志也可以正常调用 Agent但不能修改配置、不能再次分享。权限语义非常清楚。3. 用 FastAPI 和 SQLite 把最小链路跑通3.1 环境、目录与数据库表设计建议目录结构如下agentconnect-demo/ ├── main.py ├── schema.sql └── requirements.txt依赖文件保持最小fastapi uvicorn安装命令pip install -r requirements.txtschema.sql 定义用户、团队、Agent 和授权记录CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, name TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS teams ( id TEXT PRIMARY KEY, name TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS team_members ( team_id TEXT NOT NULL REFERENCES teams(id), user_id TEXT NOT NULL REFERENCES users(id), PRIMARY KEY (team_id, user_id) ); CREATE TABLE IF NOT EXISTS agents ( id TEXT PRIMARY KEY, name TEXT NOT NULL, owner_id TEXT NOT NULL REFERENCES users(id), config TEXT NOT NULL DEFAULT {}, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS agent_permissions ( id TEXT PRIMARY KEY, agent_id TEXT NOT NULL REFERENCES agents(id), grantee_type TEXT NOT NULL CHECK(grantee_type IN (user, team)), grantee_id TEXT NOT NULL, action TEXT NOT NULL CHECK(action IN (invoke, view_log, manage_config, share)), created_at TEXT NOT NULL, UNIQUE(agent_id, grantee_type, grantee_id, action) );关键设计说明agent_permissions用grantee_type区分授权给用户还是团队避免拆成两张表。grantee_id在用户级存储用户 ID在团队级存储团队 ID。UNIQUE约束保证同一对象不会重复授权同一动作。action使用 CHECK 约束限定合法动作防止脏数据进入系统。3.2 登录接口演示项目如何获得用户身份演示项目使用确定性用户 IDuser-加小写用户名。这样做是为了让 curl 示例里的 ID 可预测。生产环境必须替换为 OIDC、SSO 或企业内部身份系统的令牌。import json import sqlite3 import uuid from datetime import datetime, timezone from typing import Literal, Optional from fastapi import Depends, FastAPI, Header, HTTPException from pydantic import BaseModel DB_PATH agentconnect.db def now(): return datetime.now(timezone.utc).isoformat() def get_db(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row try: yield conn finally: conn.close() def init_db(): conn sqlite3.connect(DB_PATH) conn.executescript(open(schema.sql, encodingutf-8).read()) conn.execute(INSERT OR IGNORE INTO users(id, name) VALUES (user-alice, alice), (user-bob, bob)) conn.execute(INSERT OR IGNORE INTO teams(id, name) VALUES (team-ops, ops)) conn.execute(INSERT OR IGNORE INTO team_members(
返回列表