import asyncio
from kontext import Broker  # 开源凭证代理中间件

async def agent_with_safe_credentials(user_id: str, task: str):
    """每个 Agent 会话独立凭证作用域，运行时注入，用完即销毁"""

    broker = Broker(app_name="prod-agents")

    # 凭证不在这里——Agent 只拿到临时的、作用域受限的 token
    session = await broker.get_session(user_id)

    # LLM 客户端从代理获取，从不接触原始 Key
    llm_client = session.get_llm("openai")

    # 数据库访问也是临时的、限时的 token
    db_token = session.get_credential("postgres")

    result = await llm_client.complete(task, tools=db_token.tools())

    # 用户登出时自动吊销所有 Token
    await broker.revoke_session(user_id)
    return result

# 团队成员离职？Agent 被攻破？一键吊销该用户所有会话凭证
await broker.emergency_revoke(user_id="disgruntled_employee")

import asyncio
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
import os

AGENT_SANDBOX = Path("/tmp/ai_agents_sandbox")
AGENT_SANDBOX.mkdir(exist_ok=True)

async def run_sandboxed_agent(agent_id: int, task: str) -> dict:
    """每个 Agent 有自己独立的文件系统空间和工作目录"""

    workdir = AGENT_SANDBOX / f"agent_{agent_id:03d}"
    workdir.mkdir(exist_ok=True)

    # 构建隔离沙箱
    sandbox_config = {
        "workspace_root": str(workdir),
        "allowed_paths": [str(workdir)],      # 只能读写自己的目录
        "env_vars": {**os.environ, "AGENT_ID": str(agent_id)},
        "max_file_ops": 200,                   # 防止异常 Agent 读写过多文件
    }

    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        ProcessPoolExecutor(max_workers=1),
        execute_agent_in_sandbox,
        agent_id, task, sandbox_config
    )
    return result

async def batch_refactor(module_tasks: list[tuple[int, str]]):
    """10 个 Agent 同时开工，互不干扰"""
    results = await asyncio.gather(
        *[run_sandboxed_agent(aid, task) for aid, task in module_tasks],
        return_exceptions=True
    )
    return results

# 场景：10 个模块同时重构
tasks = [
    (0, "重构 auth 模块为 JWT RS256"),
    (1, "重构 payment 模块对接 Stripe v3"),
    (2, "重构日志模块为结构化 JSON"),
    # ... 同时跑，互不干扰
]
results = asyncio.run(batch_refactor(tasks))

// mcp_audit.js — 扫描所有已安装 MCP Server 的权限清单
import { readFileSync } from 'fs';

const DANGEROUS_PERMISSIONS = [
  'filesystem:write',    // 写任意文件
  'network:all',          // 任意网络请求
  'exec:run',             // 执行任意命令
  'credential:read',      // 读取存储凭证
];

function auditMcpServer(manifestPath) {
  const manifest = JSON.parse(readFileSync(manifestPath, 'utf-8'));
  const issues = [];

  for (const tool of manifest.tools || []) {
    const perms = tool.permissions || [];
    const danger = perms.filter(p => DANGEROUS_PERMISSIONS.includes(p));

    if (danger.length > 0) {
      issues.push({
        tool: tool.name,
        dangerous: danger,
        recommendation: suggest_minimal_permissions(tool)
      });
    }
  }

  return issues;
}

// 批量审计：npm install 后自动运行
// $ node mcp_audit.js ./node_modules/*/mcp-manifest.json

import time
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class FlowSession:
    """专注会话 + 自动上下文重置机制"""
    session_id: str
    started_at: float = field(default_factory=time.time)
    token_budget: int = 8000          # 每会话 Token 上限
    checkpoint_interval: int = 2000    # 每 2000 Token 保存一次
    checkpoints: list = field(default_factory=list)

    def should_save_checkpoint(self, tokens_used: int) -> bool:
        return tokens_used % self.checkpoint_interval == 0

    def needs_reset(self) -> bool:
        """90 分钟是经过验证的人类心流周期——对 AI Agent 同样适用"""
        return time.time() - self.started_at > 90 * 60

async def agent_flow_session(agent, task: str, session: FlowSession):
    """Agent 在专注会话中运行，达到阈值后自动重置状态"""

    tokens_consumed = 0

    while tokens_consumed < session.token_budget:
        chunk = await agent.run(task, max_tokens=1000)
        tokens_consumed += chunk.token_count

        if session.should_save_checkpoint(tokens_consumed):
            session.checkpoints.append(agent.capture_state())
            print(f"[{session.session_id}] Checkpoint saved at {tokens_consumed} tokens")

        # 时间到 or Token 用完 → 重置 Agent 状态
        if session.needs_reset() or session.token_budget - tokens_consumed < 500:
            print(f"[{session.session_id}] Flow cycle complete. Resetting agent state...")
            await save_checkpoint(session.session_id, session.checkpoints[-1])
            agent.reset()  # 清空上下文，重新开始
            session.started_at = time.time()  # 新周期
            session.checkpoints.clear()

    return session.checkpoints

from enum import Enum
from typing import Literal

class TaskComplexity(Enum):
    TRIVIAL    = ("gpt-4o-mini", 0.001)   # 每千 Token 成本（美元）
    STANDARD  = ("gpt-4o", 0.015)
    COMPLEX   = ("claude-sonnet-4", 0.015)
    EXPERT    = ("claude-opus-4", 0.075)

def assess_complexity(task: str) -> TaskComplexity:
    """简单关键词匹配 — 生产环境建议换成 LLM 分类器"""
    t = task.lower()

    if any(k in t for k in ["typo", "rename", "format", "lint", "单行修改"]):
        return TaskComplexity.TRIVIAL
    if any(k in t for k in ["refactor", "重构", "redesign", "重设计"]):
        return TaskComplexity.COMPLEX
    if any(k in t for k in ["security", "安全", "audit", "审计", "compliance"]):
        return TaskComplexity.EXPERT
    return TaskComplexity.STANDARD

def execute_routed(task: str, router_llm) -> dict:
    level = assess_complexity(task)
    model, cost = level.value

    result = router_llm.complete(task, model=model)

    return {
        "model": model,
        "cost_per_1k": cost,
        "complexity": level.name,
        "result": result
    }

# 效果：60-90% 的 Token 成本降低，响应速度提升 3-5 倍