方法介绍
MetadataClient 是 V3 管理面的 Python 客户端，用于管理用户、团队、Agent、Task、资产、权限、知识库等元数据资源。共 59 个方法，覆盖 14 组资源类型。导入
from tencentdb_agent_memory.v3 import MetadataClient, AsyncMetadataClient
构造
from tencentdb_agent_memory.v3 import MetadataClientmetadata = MetadataClient(endpoint="https://memory.tdai.tencentyun.com",api_key="sk-xxxxxxxxxxxxxxxx",service_id="tdai-mem-xxxxxxxx",user_key="sk-mem-...",timeout=30,verify=False,)
参数名 | 类型 | 必填 | 描述说明 |
endpoint | str | 是 | Memory 服务接入地址 |
api_key | str | 是 | API Key，格式 sk-... |
service_id | str | 是 | 实例 ID，如 tdai-mem-xxxxxxxx。使用自定义传输通道构造时可省略 |
user_key | str | 否 | 用户密钥。调用 create_user、delete_users 等需要用户密钥鉴权的接口时使用 |
timeout | float | 否 | 请求超时时间（秒），默认 30 |
verify | bool | 否 | 是否验证 SSL 证书，默认 False |
stub | Stub | 否 | 自定义网络传输通道，用于测试场景（如注入 mock） |
用户管理
管理 Memory 实例下的用户账号。
# 创建用户resp = metadata.create_user({"username": "alice","team_id": "team-abc",})# 按 ID 或条件查询user = metadata.get_user("user-123")user = metadata.get_user({"username": "alice", "team_id": "team-abc"})# 删除用户resp = metadata.delete_users(["user-123", "user-456"])# 按团队或条件列出用户resp = metadata.list_users("team-abc")resp = metadata.list_users({"team_id": "team-abc","limit": 20,"offset": 0,})
方法 | 参数 | 描述 |
create_user(payload) | Dict | 创建用户，含 username、team_id 等字段 |
get_user(user_id_or_filter) | str 或 Dict | 按 ID 或过滤条件查询单个用户 |
delete_users(user_ids) | List[str] | 批量删除用户 |
list_users(team_id_or_request, *, pagination) | str 或 Dict, Dict | 按团队 ID 或条件查询用户列表 |
用户密钥管理
管理用户的 API 密钥。
# 创建密钥resp = metadata.create_user_key({"user_id": "user-123", "name": "default"})print(resp["key_id"])# 列出密钥resp = metadata.list_user_keys("user-123")# 查询单个密钥resp = metadata.get_user_key("key-abc")# 吊销密钥resp = metadata.revoke_user_key("key-abc")# 更新密钥resp = metadata.update_user_key({"key_id": "key-abc", "name": "renamed"})
方法 | 参数 | 描述 |
create_user_key(payload) | Dict | 为用户创建新密钥 |
list_user_keys(user_id) | str 或 Dict | 列出用户密钥 |
get_user_key(key_id) | str | 查询单个密钥信息 |
revoke_user_key(key_id) | str | 吊销密钥 |
update_user_key(payload) | Dict | 更新密钥信息 |
团队管理
管理团队资源。
# 创建团队resp = metadata.create_team({"name": "Platform Team"})# 查询团队resp = metadata.get_team("team-abc")# 更新团队resp = metadata.update_team({"team_id": "team-abc", "name": "New Name"})# 删除团队resp = metadata.delete_teams(["team-abc"])# 列出用户所属团队resp = metadata.list_teams("user-123")for team in resp["items"]:print(team["name"])
方法 | 参数 | 描述 |
create_team(payload) | Dict | 创建团队 |
get_team(team_id) | str | 按 ID 查询团队 |
update_team(payload) | Dict | 更新团队信息 |
delete_teams(team_ids) | List[str] | 批量删除团队 |
list_teams(user_id) | str | 列出用户所属团队 |
团队成员管理
管理团队成员关系。
# 添加成员resp = metadata.add_team_member({"team_id": "team-abc","user_id": "user-123","role": "member",})# 列出成员resp = metadata.list_team_members("team-abc", pagination={"limit": 50, "offset": 0})for m in resp["items"]:print(m["user_id"], m["role"])# 查询单个成员resp = metadata.get_team_member("team-abc", "user-123")# 移除成员resp = metadata.remove_team_member("team-abc", "user-123")
方法 | 参数 | 描述 |
add_team_member(payload) | Dict | 添加成员到团队 |
remove_team_member(team_id, user_id) | str, str | 从团队移除成员 |
list_team_members(team_id, *, pagination) | str, Dict | 列出团队成员 |
get_team_member(team_id, user_id) | str, str | 查询单个成员信息 |
Agent 管理
管理 Agent 资源。
# 创建 Agentresp = metadata.create_agent({"name": "sql-helper","team_id": "team-abc","model": "gpt-4o",})# 查询 Agentresp = metadata.get_agent("agt-xyz789")# 更新 Agentresp = metadata.update_agent({"agent_id": "agt-xyz789", "model": "gpt-4.1"})# 列出团队下 Agentresp = metadata.list_agents({"team_id": "team-abc"})for agent in resp["items"]:print(agent["name"])# 删除 Agentresp = metadata.delete_agents(["agt-xyz789"])# 归档 Agent（软删除）resp = metadata.archive_agent("agt-xyz789")
方法 | 参数 | 描述 |
create_agent(payload) | Dict | 创建 Agent |
get_agent(agent_id) | str | 按 ID 查询 Agent |
update_agent(payload) | Dict | 更新 Agent 信息 |
delete_agents(agent_ids) | List[str] | 批量删除 Agent |
list_agents(payload) | Dict | 按条件列表查询 |
archive_agent(agent_id) | str | 归档 Agent（软删除，可恢复） |
Task 管理
管理 Task 资源。
# 创建 Taskresp = metadata.create_task({"name": "Q3 数据迁移", "team_id": "team-abc"})# 查询 Taskresp = metadata.get_task("task-xyz")# 更新 Taskresp = metadata.update_task({"task_id": "task-xyz", "name": "Q3 数据迁移 v2"})# 列出 Taskresp = metadata.list_tasks("team-abc", status="active", pagination={"limit": 20})# 删除 Taskresp = metadata.delete_tasks(["task-xyz"])# 归档 Taskresp = metadata.archive_task("task-xyz")
方法 | 参数 | 描述 |
create_task(payload) | Dict | 创建 Task |
get_task(task_id) | str | 按 ID 查询 Task |
update_task(payload) | Dict | 更新 Task 信息 |
delete_tasks(task_ids) | List[str] | 批量删除 Task |
list_tasks(team_id, *, status, pagination) | str, str, Dict | 按团队和状态列表查询 |
archive_task(task_id) | str | 归档 Task（软删除） |
Task-Agent 关联
管理 Task 与 Agent 的绑定关系。
# 绑定 Agent 到 Taskresp = metadata.link_task_agent("task-xyz", "agt-abc", role_in_task="executor")# 列出 Task 关联的 Agentresp = metadata.list_task_agents("task-xyz", pagination={"limit": 50})# 解除绑定resp = metadata.unlink_task_agent("task-xyz", "agt-abc")
方法 | 参数 | 描述 |
link_task_agent(task_id, agent_id, *, role_in_task) | str, str, str | 绑定 Agent 到 Task |
unlink_task_agent(task_id, agent_id) | str, str | 解除 Task 与 Agent 的绑定 |
list_task_agents(task_id, *, pagination) | str, Dict | 列出 Task 关联的 Agent |
参与日志
管理 Agent 参与 Task 的执行日志。
# 追加参与日志resp = metadata.append_participation_log({"task_id": "task-xyz","agent_id": "agt-abc","event": "completed",})# 查询参与日志resp = metadata.list_participation_logs({"task_id": "task-xyz","limit": 50,})
方法 | 参数 | 描述 |
append_participation_log(payload) | Dict | 追加参与日志 |
list_participation_logs(payload) | Dict | 查询参与日志 |
资产管理
管理共享资产资源。
# 创建资产resp = metadata.create_asset({"name": "数据字典 v2.0","team_id": "team-abc","type": "document",})# 查询资产resp = metadata.get_asset("asset-456")# 更新资产resp = metadata.update_asset({"asset_id": "asset-456", "name": "数据字典 v2.1"})# 列出资产resp = metadata.list_assets({"team_id": "team-abc"})# 查询用户可访问的资产resp = metadata.list_accessible_assets({"user_id": "user-123", "limit": 50})# 标记最近使用resp = metadata.touch_asset_usage("asset-456")# 删除资产resp = metadata.delete_assets(["asset-456"])
方法 | 参数 | 描述 |
create_asset(payload) | Dict | 创建资产 |
get_asset(asset_id) | str | 按 ID 查询资产 |
update_asset(payload) | Dict | 更新资产信息 |
delete_assets(asset_ids) | List[str] | 批量删除资产 |
list_assets(payload) | Dict | 按条件列表查询 |
list_accessible_assets(payload) | Dict | 查询用户可访问的资产 |
touch_asset_usage(asset_id) | str | 标记资产最近使用时间 |
Agent 固定资产
管理 Agent 的固定资产绑定。
# 全量设置固定资产resp = metadata.set_agent_fixed_assets("agt-xyz789", [{"asset_id": "asset-123", "priority": 1},{"asset_id": "asset-456", "priority": 2},])# 列出 Agent 固定资产resp = metadata.list_agent_fixed_assets("agt-xyz789", pagination={"limit": 50})# 列出固定资产含详情resp = metadata.list_agent_fixed_assets_with_detail({"agent_id": "agt-xyz789"})# 按 Agent 汇总固定资产resp = metadata.summarize_agent_fixed_assets_by_agents({"team_id": "team-abc"})
方法 | 参数 | 描述 |
set_agent_fixed_assets(agent_id, bindings) | str, List[Dict] | 全量设置 Agent 固定资产绑定 |
list_agent_fixed_assets(agent_id, *, pagination) | str, Dict | 列出固定资产 |
list_agent_fixed_assets_with_detail(payload) | Dict | 列出固定资产含详情 |
summarize_agent_fixed_assets_by_agents(payload) | Dict | 按 Agent 汇总固定资产 |
访问控制（ACL）
管理资产级别的访问权限。
# 授权resp = metadata.grant_acl({"asset_id": "asset-456","user_id": "user-123","permission": "read",})# 检查权限resp = metadata.check_acl({"asset_id": "asset-456","user_id": "user-123",})print(resp["allowed"])# 列出资产 ACLresp = metadata.list_acl("asset-456", pagination={"limit": 50})# 撤销授权resp = metadata.revoke_acl("acl-001")
方法 | 参数 | 描述 |
grant_acl(payload) | Dict | 授予用户对资产的访问权限 |
revoke_acl(acl_id) | str | 撤销已有授权 |
list_acl(asset_id, *, pagination) | str, Dict | 列出资产的访问控制列表 |
check_acl(payload) | Dict | 检查用户对资产的权限 |
鉴权
验证用户密钥有效性。
resp = metadata.verify_auth("sk-mem-...")print(resp["valid"])
方法 | 参数 | 描述 |
verify_auth(user_key) | str | 验证用户密钥是否有效 |
配置与配额
管理实例配额和用户配置。
# 获取实例配额resp = metadata.get_instance_quota()print(resp["max_users"], resp["max_teams"])# 获取用户配置resp = metadata.get_user_config({"user_id": "user-123", "key": "language"})# 设置用户配置resp = metadata.set_user_config({"user_id": "user-123","key": "language","value": "zh-CN",})
方法 | 参数 | 描述 |
get_instance_quota() | - | 获取实例配额上限 |
get_user_config(payload) | Dict | 获取用户配置项 |
set_user_config(payload) | Dict | 设置用户配置项 |
知识库管理
管理知识库资源。
# 创建知识库resp = metadata.create_knowledge({"name": "产品文档","team_id": "team-abc",})# 查询知识库resp = metadata.get_knowledge("kb-123", team_id="team-abc")# 更新知识库resp = metadata.update_knowledge({"knowledge_id": "kb-123","name": "产品文档 v2",})# 列出知识库resp = metadata.list_knowledge({"team_id": "team-abc"})# 删除知识库resp = metadata.delete_knowledge(["kb-123"], team_id="team-abc")
方法 | 参数 | 描述 |
create_knowledge(payload) | Dict | 创建知识库 |
get_knowledge(knowledge_id, *, team_id) | str, str | 按 ID 查询知识库 |
update_knowledge(payload) | Dict | 更新知识库 |
delete_knowledge(knowledge_ids, *, team_id) | List[str], str | 删除知识库 |
list_knowledge(payload) | Dict | 列出知识库 |
上下文管理器
推荐使用上下文管理器自动管理连接生命周期：
with MetadataClient(endpoint="https://memory.tdai.tencentyun.com",api_key="sk-xxxxxxxxxxxxxxxx",service_id="tdai-mem-xxxxxxxx",) as metadata:user = metadata.create_user({"username": "alice", "team_id": "team-abc"})agents = metadata.list_agents({"team_id": "team-abc"})
退出
with 块时自动调用 close() 释放网络连接。异步版本
AsyncMetadataClient 的方法签名与 MetadataClient 完全一致，所有方法均为 async：import asynciofrom tencentdb_agent_memory.v3 import AsyncMetadataClientasync def main():async with AsyncMetadataClient(endpoint="https://memory.tdai.tencentyun.com",api_key="sk-...",service_id="tdai-mem-xxxxxxxx",) as metadata:users = await metadata.list_users("team-abc")agent = await metadata.create_agent({"name": "sql-helper","team_id": "team-abc","model": "gpt-4o",})print(f"已创建 Agent: {agent['agent_id']}")asyncio.run(main())
