loopy

Loopy 是一个基于 Python asyncio 的通用 LLM Agent 平台，提供多轮对话、项目任务、工具调用、后台执行、MCP 接入和上下文压缩。你可以通过网页或桌面界面使用，也可以在终端中操作、通过飞书机器人交互，或用 Python SDK 集成到自己的应用。

介绍

选择使用入口

UI、CLI、Gateway 在公共应用层 loopy.application 汇合，复用会话、工具、MCP 和上下文管理。核心 SDK 位于 loopy，可独立使用。

核心能力

• 持续任务：多轮会话、流式回答、后台工具执行，以及分级上下文压缩和历史回溯。
• 工具接入：命令执行、文件操作、用户提问、MCP，以及动态加载的 Python 工具。
• 共享运行时：Agent 与外部脚本访问同一组工具实例；支持批量方法、流式方法和独立 worker。
• System1：通过 State → Action 接口接入策略，记录观察、决策和动作执行回执。
• 执行可见：保存 JSONL 事件与工具结果；界面展示历史、工具调用、问题确认和压缩过程。

模型目前通过远端服务调用，支持 Anthropic 消息协议。源码运行需要 Python 3.11+；界面前端和桌面构建还需要 Node.js / npm，CI 使用 Node.js 22。发行版包含相应运行组件，Bash、Git、npx 等外部工具仍需在实际执行环境安装。

界面版

网页：启动与首次配置

在仓库根目录启动 Backend：

uv sync --project ui/backend --extra dev --frozen
uv run --project ui/backend loopy-ui

另开终端启动前端：

cd ui/frontend
npm ci
npm run dev:h5

打开开发服务器输出的地址，在“设置 → 模型配置”中填写并激活模型，再创建独立对话或导入项目目录。运行时设置用于选择 Host / WSL；MCP 设置用于接入外部工具。

Backend 默认监听 127.0.0.1:8765，数据目录为 ~/.loopy-ui。使用 LOOPY_UI_HOST、LOOPY_UI_PORT、LOOPY_UI_DATA_DIR、LOOPY_UI_AUTH_TOKEN 等环境变量覆盖配置。模型密钥可通过界面安全存储，或使用 LOOPY_UI_ANTHROPIC_API_KEY 提供。单次模型输出默认上限为 8192，可通过 LOOPY_UI_MAX_TOKENS 调整。

桌面与 Backend 连接

Windows / macOS 桌面版默认启动随包提供的 Python Backend（Sidecar），可直接操作本机项目。项目目录始终属于所连接的 Backend：连接远端时，使用的是远端目录。

远端 Backend 模式显示“backend 测试”设置；本地 Sidecar 模式隐藏该入口。桌面也可通过 LOOPY_BACKEND_URL 和 LOOPY_BACKEND_TOKEN 指定远端服务。远端服务需要 HTTPS、认证和相应 CORS；桌面文件页面的 Origin 为 null，需要在服务端明确允许。

桌面外链通过系统浏览器打开，原生桥接仅向本地应用页面提供。移动端目前已有本地对话的 Python JSON 桥接边界，但 Android / iOS / Harmony 原生 App 仍需平台工程和设备验证，详见 原生宿主说明。

仅对话部署

普通 UI Backend 默认同时支持项目和独立对话。设置 LOOPY_UI_CONVERSATION_ONLY=true 后重启 Backend，可部署仅对话服务；本机和远端部署均适用，同一份网页根据能力接口隐藏项目入口，无需单独编译。

cd ui/backend
$env:LOOPY_UI_CONVERSATION_ONLY = "true"
uv run python -m loopy_ui_backend.main

刷新网页后生效。恢复完整模式时将该变量设为 false 并重启 Backend。仅对话模式复用公共运行层的受限工具集，不加载本地命令、文件操作或 MCP 工具；项目 API 返回 403，Session 列表只展示独立对话，已有项目数据不会删除。调试时可用 LOOPY_UI_DATA_DIR 指定独立数据目录。

MCP 配置

在“设置 → MCP 配置”新增服务器：

• 本地进程（stdio）：单行填写启动命令和参数，例如 npx -y @example/mcp-server；含空格的路径或参数使用引号包裹。列表和查看弹框也单行展示。使用会话工作目录，环境变量单独配置。命令会解析为程序和参数数组，不经过 shell。
• 远程服务（HTTP）：填写 Streamable HTTP URL、可选 Header 或 Bearer 来源。HTTPS 可用于远程服务；HTTP 仅用于 loopback。
• 环境变量引用在实际执行环境解析。秘密值选择“安全存储”，只写不回显；编辑时留空保持原值，删除对应项明确清除。
• 安全存储不可用时改用环境变量引用，不会把秘密明文写入 SQLite。
• 高级设置包含启动/调用超时、工具过滤以及“连接失败时阻止会话运行”。允许列表未设置表示全部允许，显式空列表表示不暴露任何工具；禁用列表优先。

“测试连接”会启动程序或连接服务，仅进行协议初始化和工具发现，不调用模型或业务工具。查看测试环境、工作目录和工具预览；测试通过不保证每个工具的业务权限。HTTP 不自动跟随重定向，请填写最终 endpoint；loopback 直连，其他地址沿用进程代理环境。

新增配置默认禁用，保存后在列表启用；可同时启用多个服务器。会话首次执行固定配置快照，后续设置改动默认只影响尚未首次执行的会话。已有会话可在 MCP 配置页点击「重置」，于下次发送消息前加载最新配置；正在执行的任务不受影响，会话历史与其他工具保持不变。重置不清空配置，失败可检查配置后重试。外部 stdio 程序管理的登录态独立于 Loopy 配置快照。

会话、历史与执行看板

对话页支持流式回答、回答 Agent 提问、取消运行和查看工具结果。默认加载最近 30 轮，点击“加载更早的对话”继续向前翻页；待回答问题在进入会话和重连时恢复。

模型配置中的“上下文窗口”用于启用按预算自动压缩；留空时不启用水位触发，Agent 仍可指定层级压缩。界面展示压缩过程及历史版本，token 数值为本地估算。

输入框“+”菜单提供后台工具和 System1 看板。System1 展示 Agent 决策使用的观察及脚本上报的执行回执，不代表环境实时状态；决策成功也不等于动作已执行。

Backend 按会话持有工具 Runtime。删除会话、删除项目或退出 Backend 会清理对应资源；重启后未完成的调用标记为中断，不自动恢复此前的业务进程。

构建与桌面打包

网页构建在 ui/frontend 执行 npm run build:h5，输出为 ui/frontend/dist。桌面包需要先构建 Backend，再打包前端和 Electron 宿主：

# 仓库根目录
uv sync --project ui/backend --extra dev --frozen
cd ui/backend
uv run python -m PyInstaller --noconfirm loopy-ui.spec
cd ../frontend
npm ci
npm run package:windows -- --publish never
# macOS 上改用：npm run package:macos -- --publish never

安装包输出到 ui/frontend/release。npm run build:electron 只编译前端与宿主，不会单独生成 Python Backend；完整打包需执行上面的两阶段流程。build:rn / build:harmony 是前端构建目标，不表示已有可发行的手机 App。

CLI

安装与打开项目

Loopy CLI 面向本地项目，同时提供 Gateway 启动入口。首次配置模型后即可打开目录：

loopy config model
loopy open /path/to/project

源码运行：

uv sync --extra dev --frozen
uv run python -m loopy.entry config model
uv run python -m loopy.entry open /path/to/project

Python 统一发行包为 loopy-agent-fs（需要 Python 3.11+）。发布到 PyPI 后可通过 python -m pip install loopy-agent-fs 安装 SDK、CLI 和飞书 Gateway；当前可使用源码安装或 uv build 构建的 wheel。建议使用独立虚拟环境，避免与其他提供 loopy 模块的发行包冲突。CLI 在本地运行 Backend，Bash、npx 等外部程序仍需在运行环境中安装。

常用命令

loopy config model --action list
loopy config model --action activate --id <model-id>
loopy config runtime                         # 查看运行配置
loopy config runtime --file runtime.json     # 更新 Host / WSL 配置
loopy config mcp --file mcp.json             # 新建，格式与 MCP 配置接口一致
loopy config mcp --action enable --id <server-id>
loopy config mcp --action test --id <server-id>
loopy projects list
loopy sessions list --project <project-id>
loopy chat --session <session-id>
loopy run --workspace /path/to/project --prompt "分析项目结构" --json

open／无参数 chat 默认创建新 session；项目按目录复用。--json 输出 JSONL 事件；遇到需要人工回答的问题时输出问题并取消任务、返回非零退出码，避免脚本无限等待。交互模式可直接回答问题，/exit 退出。Ctrl+C 取消并关闭本次运行。

数据与凭据

默认数据根目录为 ~/.loopy，通过顶层 --data-dir 或 LOOPY_DATA_DIR 选择独立实例。一个数据目录同时只允许一个进程管理运行。模型凭据使用系统 keyring；无可用 keyring 的服务环境使用 LOOPY_ANTHROPIC_API_KEY，不写明文配置。现有 UI 继续使用 ~/.loopy-ui，不会自动搬迁或合并历史。

Gateway

飞书机器人接入

Gateway 与交互 CLI 共用 loopy 发行包，无需安装另一套命令。目前仅支持飞书，使用长连接接收消息。

1. 在飞书开放平台创建企业自建应用，启用机器人，申请消息接收与机器人发送消息所需权限，发布可用版本。
2. 配置 im.message.receive_v1 事件并选择长连接接收；本期不要求公网 webhook。
3. 准备 App ID、App Secret、机器人的 open_id、允许访问的用户 open_id 和群 chat_id。
4. 在本机先完成 loopy config model，再运行 loopy config gateway。App Secret 存系统 keyring，也可通过 LOOPY_FEISHU_APP_SECRET 提供。
5. 运行 loopy gateway --workspace /path/to/project，或省略 workspace 使用独立对话。将机器人加入允许的群，授权用户 @机器人并发送文本；私聊也会检查用户允许名单。

同一飞书对话／群映射一个 session，多群可以共用启动时指定的目录。共享项目忙碌时拒绝新任务，不排队。日志记录消息、session、turn 与投递标识；不记录凭据或原始消息正文。群中可使用 /answer <question-id> <answer> 和 /cancel <turn-id>，仅原任务发送者可操作。

网络错误只重试发送，不重新运行 Agent。结果不确定的发送会暂停：停止 Gateway 后，用 loopy deliveries --action list 查询，人工核对飞书消息后用 --action acknowledge --id <outbox-id> 确认已送达，或 --action retry --id <outbox-id> 显式重试，再启动 Gateway。未核对前重试可能产生重复回复。

常驻 Gateway 与交互 CLI 如需并行，请分别指定 --data-dir 并配置各自模型／MCP。微信、钉钉不在本期范围。

SDK

选择接入层级

安装与模型配置

Python SDK 需要 Python 3.11 或更高版本。调用模型时需要 Anthropic API Key，或兼容 Anthropic 消息协议的服务配置。

在仓库根目录安装开发环境：

uv sync --extra dev

也可以使用 pip 安装当前源码：

python -m pip install -e ".[dev]"

SDK 默认读取 ANTHROPIC_API_KEY。以下示例通过 LOOPY_MODEL 指定模型名称：

export ANTHROPIC_API_KEY="your-api-key"
export LOOPY_MODEL="your-model-name"

Windows PowerShell：

$env:ANTHROPIC_API_KEY = "your-api-key"
$env:LOOPY_MODEL = "your-model-name"

也可向 Agent 显式传入 api_key 和 base_url。工具运行时本身不依赖模型凭据，可以独立用于脚本调用与测试。文本搜索可选安装 ripgrep；未安装时会回退到 grep 或 Python 实现。

公共应用层：复用项目与会话管理

统一发行包默认包含公共应用层、CLI、Gateway 与 MCP；源码安装可执行 python -m pip install -e .。底层 Agent 使用 ANTHROPIC_API_KEY；公共应用层使用 LOOPY_ANTHROPIC_API_KEY 或已保存的模型配置，两者不要混淆。

import asyncio
from loopy.application import LoopyApplication, Settings


async def main():
    # 先通过 CLI 配置模型，或设置 LOOPY_ANTHROPIC_API_KEY。
    async with LoopyApplication(Settings()) as app:
        project = app.workspaces.open(".")
        session = await app.sessions.create(workspace_id=project["id"])
        turn = await session.submit("分析项目结构", request_id="analysis-001")
        print(await turn.result())


asyncio.run(main())

应用层默认使用 ~/.loopy，同一数据目录只允许一个管理进程。CLI / Gateway 已占用该目录时，使用独立目录，或通过远端 SDK 连接运行中的 UI Backend。任务可能向用户提问，交互应用应查询 pending_questions(session_id) 并使用 answer_question(...) 回答。

远端应用层：连接已有 Backend

import asyncio
import os
from loopy.application import LoopyClient


async def main():
    async with LoopyClient.connect(
        os.environ["LOOPY_BACKEND_URL"],
        credentials=os.environ["LOOPY_BACKEND_TOKEN"],
    ) as client:
        session = await client.sessions.create(title="SDK conversation")
        turn = await session.submit("你好", request_id="hello-001")
        async for event in turn.events():
            print(event)
        print(await turn.result())


asyncio.run(main())

远端 SDK 复用服务端配置与数据，远程地址要求 HTTPS，loopback 可使用 HTTP。需要人工确认的任务应在消费事件时处理 question.required，通过 client.answer_question(question_id, answer) 回答；仅打印事件不会解除等待。

底层 Agent：快速开始

将下面的代码保存为 main.py，然后执行 uv run python main.py：

import asyncio
import os
from pathlib import Path

from loopy import Agent, AskUserQuestion, Bash, Grep, ToolRuntime, Write


async def main() -> None:
    workspace = Path("workspace")
    workspace.mkdir(parents=True, exist_ok=True)

    async with ToolRuntime(
        session_id="session-1",
        working_dir=workspace,
        tools=[Bash(), Write(), Grep(), AskUserQuestion()],
    ) as runtime:
        agent = Agent(
            name="assistant",
            system_prompt="你是一个可靠的项目助手，使用工具完成用户的任务。",
            model=os.environ["LOOPY_MODEL"],
            working_dir=workspace,
            tool_runtime=runtime,
        )
        try:
            session = agent.new_session()
            response = await session.run("查看工作目录，说明有哪些文件。")
            print(response.text)

            response = await session.run("把刚才的说明写入 summary.md。")
            print(response.text)
        finally:
            await agent.aclose()


if __name__ == "__main__":
    asyncio.run(main())

常用配置：

Agent Loop 不限制工具执行轮数，直到模型结束本轮、用户取消或发生错误。遇到 max_tokens 时，Loop 保留已生成的正文或思考块，加入内部继续指令，并以相同输出预算发起新请求。输出恢复单独计数，工具轮完成后重置，过程记录为 llm_continuation 事件。连续正文合并到 FinalResponse.text。响应为空时明确报错；包含可能截断的工具调用时，丢弃整次响应，不执行其中任何工具，要求模型缩小操作后重新生成。两种恢复共用次数上限，截断参数不进入正式历史；此前成功的工具调用不会重放。续写依赖模型服务支持回传已有内容，不能保证模型完全不重复。

working_dir 是执行与文件管理的基准目录。Write、Grep 会检查路径范围；Bash 执行的是本机 shell 命令，工作目录并不构成进程沙箱。

会话与生命周期

session.run(query) 可以重复调用，session.history() 返回当前消息历史，session.session_id 标识会话。读取落盘记录和恢复可写会话使用不同入口：

# 在 agent 尚未关闭时执行。
restored = agent.get_session(session.session_id)
history = restored.history()  # get_session 返回只读会话

resumed = agent.resume_session(session.session_id, history)
response = await resumed.run("继续处理这个任务。")

一个 Agent 绑定一个 Runtime 和可写会话 ID；显式传给 new_session() 或 resume_session() 的 ID 必须与 Runtime 一致。同一 Runtime 不支持并发执行多个模型轮次。需要并发处理独立会话时，分别创建 Agent 和运行时。历史恢复只恢复消息，不重建此前的进程、流或工具业务状态。

Runtime 始终拥有注册表和工具实例，Agent 借用它们。创建者负责关闭 Runtime，建议像快速开始示例一样使用 async with。

关闭 Agent 后，可以让另一个 Agent 借用同一 Runtime，继续使用原有工具状态。工具需要释放连接、文件或业务资源时，实现 aclose()。

工具接口

内置工具

前四项通过 ToolRuntime(tools=[...]) 注册。runtime.enable_management() 注册 LoadTool 和 ToolControl，UI 默认启用这两个管理工具。

单方法工具

继承 BatchTool，提供描述、参数 Schema、validate() 和 run()：

from loopy import BatchTool, ToolContext, ToolResult


class Echo(BatchTool):
    name = "Echo"
    description = "返回输入文本"
    input_schema = {
        "type": "object",
        "properties": {"text": {"type": "string"}},
        "required": ["text"],
    }

    async def validate(self, ctx: ToolContext, args: dict) -> dict:
        if not isinstance(args.get("text"), str):
            raise ValueError("text must be a string")
        return args

    async def run(self, ctx: ToolContext, args: dict) -> ToolResult:
        return ToolResult(call_id=ctx.call_id, content=args["text"])

将 Echo() 加入 Runtime 的 tools 列表，或调用 runtime.register(Echo()) 完成注册。模型使用扁平参数，例如 {"text": "hello"}。

持续输出的单方法工具继承 StreamingTool，实现异步生成器 stream(ctx, args)，逐次 yield ToolResult(...)。每个结果的 call_id 必须与本次 ctx.call_id 一致。

多方法工具

一个工具类可以同时提供批量方法和流式方法。下面是一个独立可用的计数器工具；也可将代码保存为工作目录中的 counter_tool.py，用于动态加载：

import asyncio

from loopy import Tool, ToolMethod, ToolResult


class Counter(Tool):
    name = "Counter"
    description = "按会话和业务标识维护计数，并持续输出当前值"

    def __init__(self):
        self.values = {}

    def methods(self):
        schema = {"type": "object", "properties": {}}
        return [
            ToolMethod("increment", "计数加一", schema, mode="batch"),
            ToolMethod("watch", "持续输出计数", schema, mode="stream"),
        ]

    async def validate(self, ctx, args):
        if ctx.union_id is None:
            raise ValueError("Counter requires a union_id")
        return args

    async def run(self, ctx, args):
        key = (ctx.session_id, ctx.union_id)
        value = self.values.get(key, 0) + 1
        self.values[key] = value
        return ToolResult(ctx.call_id, str(value), meta={"value": value})

    async def stream(self, ctx, args):
        key = (ctx.session_id, ctx.union_id)
        while True:
            yield ToolResult(ctx.call_id, str(self.values.get(key, 0)))
            await asyncio.sleep(0.1)


def create_tool():
    return Counter()

ToolMethod 描述方法名称、参数和执行模式；ToolContext 携带某次实际调用的上下文，ctx.method 对应所选 ToolMethod.name。框架依据方法的 mode 调用 run() 或 stream()，同类中有多个同模式方法时，由实现根据 ctx.method 分派业务逻辑。

模型调用多方法工具时使用统一参数结构：

{"method": "increment", "union_id": "counter-a", "arguments": {}, "is_wait": true}

框架允许 union_id=None，不据此创建、隔离或自动释放资源。是否需要业务标识及如何管理关联状态，由工具定义；上面的计数器选择要求显式标识。

共享运行时与脚本调用

ToolRuntime 组织注册表、调度器和本地工具服务，工具统一通过其构造参数或 register() 注册。模型调用和脚本 RPC 最终进入同一个调度器，访问同一组工具实例。

下面的宿主代码使用内置 Write 工具，让一个独立脚本通过 RPC 写入文件。先将下一段脚本保存到 workspace/worker.py，再运行宿主代码：

import asyncio
from pathlib import Path

from loopy import Bash, ToolRuntime, Write


async def main():
    workspace = Path("workspace")
    workspace.mkdir(exist_ok=True)
    async with ToolRuntime(
        session_id="shared-session", working_dir=workspace,
        tools=[Bash(), Write()],
    ) as runtime:
        runtime.enable_management()
        await runtime.start_server()
        reply = await runtime.call(
            "Bash", arguments={"command": "python worker.py"}, is_wait=False,
        )
        while reply.status == "running":
            await asyncio.sleep(0.1)
            reply = runtime.status(reply.call_id)
        print(reply.status, reply.content)


asyncio.run(main())

worker.py 使用异步客户端访问宿主中的工具：

import asyncio
from loopy.client import ToolClient


async def main():
    async with ToolClient.from_env() as client:
        reply = await client.call(
            "Write", arguments={"path": "from-script.txt", "content": "hello"},
        )
        if reply.is_error:
            raise RuntimeError(reply.content)
        print(reply.content)


asyncio.run(main())

运行脚本的 Python 环境需要安装 loopy。Bash 为子进程注入 LOOPY_TOOL_ENDPOINT 和 LOOPY_TOOL_TOKEN，不修改宿主的全局环境变量。

需要模型参与时，将同一个运行时传给 Agent(..., tool_runtime=runtime)；会话默认使用该运行时的 session_id，工作目录必须一致。Session.run() 会按需启动工具服务。宿主必须保持事件循环和 Runtime 存活，后台任务才能继续工作。

执行规则：

• 批量方法默认等待完成；is_wait=False 返回后台回执。
• 流式方法始终在后台执行，返回 call_id 和 result_dir。
• ToolClient.status(call_id)、ToolClient.cancel(call_id) 均为异步方法；宿主的 runtime.status(call_id) 是同步方法。
• 取消后台 Bash 会撤销脚本凭据、停止进程树，并清理该脚本发起的活动调用。其他独立调用与工具业务状态保留，Agent 可以继续操作同一实例。

模型结束回答后不会自动轮询或唤醒。需要持续判断条件时，由脚本实现，或由宿主安排后续模型轮次。

System1 与环境接入

System 1 采用 State → Action 接口，负责根据任务状态选择动作。环境观测、状态提取、动作执行和反馈循环由 Agent 编写的 Python 脚本完成，Agent Loop 本身不需要接收 env。规则、RL policy 和模型适配器可以实现同一个接口：

from loopy.system1 import Action, State, System1, System1Tool


class LocalPolicy(System1):
    async def step(self, state: State) -> Action:
        return Action("action", {"direction": state.state["direction"]}, state.state_id)


runtime.register(System1Tool(LocalPolicy(), step_timeout=1.0, max_steps=64))

State.state 和 Action.action 支持任意合法 JSON 值；任务不必共享 observation 字段。State 还可携带 instruction、state_id 和 candidates（每项包含 id、value、可选 description）。策略应原样回传 state_id；提供候选时返回动作必须匹配候选值。Action("defer", state_id=...) 表示无法决定，不能当作任务结束。不要用动作的真假值判断是否执行：null、false、0 都可能是有效动作值。

生成脚本通过现有 ToolClient 调用宿主工具：

import json
from loopy import ToolClient


async def decide_once():
    async with ToolClient.from_env() as client:
        reply = await client.call(
            "System1", method="step", union_id="episode-1",
            arguments={"state": {"direction": "left"}, "state_id": "frame-1"},
        )
        if reply.is_error or reply.status != "succeeded":
            raise RuntimeError(reply.content)
        return json.loads(reply.content)

• step 是 batch 方法，每次返回一个动作；脚本按“新观测 → step → 环境动作 → 新观测”反复调用，实现闭环。
• steps 是 stream 方法，接收 {"states": [...]}，按输入顺序输出带 index 的动作。它评估已经提供的状态，不执行动作，也不预测后续环境反馈。
• Python 进程内也可使用 policy.steps(async_state_source)；下一份状态在调用方消费上一个动作后才被请求，支持按反馈逐步供给状态。
• 参数非法、输出不符合候选或 state_id、推理超时都会使调用失败。批量中途失败保留已有分片，不再处理后续状态。defer 是正常决策结果，与服务故障不同。
• 每个工具实例拥有其策略，Runtime 关闭时调用策略的 aclose()。同一个可变策略实例不要交给多个工具持有；任务记忆由脚本维护。

环境接入继续复用现有 Tool、ToolClient 和脚本。同步环境按观察、决策、执行、反馈推进；异步环境自行推进，脚本每次取最新可用观察，并在提交动作前检查时效。Self-play 使用不同策略实例和席位，外部多人环境自行处理轮次、权限及人类玩家界面。具体脚本约定可通过 ToolControl(action="guide", topic="system1") 查看。

step 和每个 steps 输入可选提供 context，记录 environment_id、participant_id、episode_id、policy_version、tick、observed_at、received_at、observation_age_ms。只填写已知信息；它们用于诊断和展示，不传给策略，也不改变业务 state。时间戳需带时区，年龄使用本地单调时钟估算，不能相减不同机器的时钟。

策略完成后，可通过 report_execution 上报实际环境执行证据：

# reply 是之前 step 的回执；episode 与该次 step 的 union_id 一致。
# 仅在环境明确确认执行后上报 executed，不能由策略成功推断。
reported = await client.call("System1", method="report_execution", union_id=episode,
    arguments={"decision_call_id": reply.call_id, "request_id": request_id,
               "status": "executed", "evidence": "environment"})

支持 submitted / accepted / executed / rejected / failed / unknown / skipped；依据为 client / environment / observation。回执沿用原决策的调用者和 union_id，同一动作始终使用同一 request_id；steps 还需提供对应 index。超时或取消不能证明动作没有执行，不自动重试非幂等动作。上报只关联工具实例最近 1024 个成功决策，重启或引用过期后不再接受；上报失败不代表动作失败。defer 不对应环境动作。

使用 Jev 时安装可选依赖：

pip install -e .

from loopy.system1 import System1Tool
from loopy.system1.jev import JevSystem1


def create_tool():
    return System1Tool(JevSystem1.from_env())

这个工厂可以通过现有 LoadTool 加载；不要求修改 UI 或自动注册内置工具。JevSystem1 使用 TypeSafe HTTP API 的 Choice 协议，将候选 ID 映射回本地动作值。支持 1–254 个候选，额外保留一个拒绝决策选项；无候选的连续动作生成不属于该适配器的能力范围。

凭据读取 TYPESAFE_API_KEY。以下为 Loopy 适配器约定的可选环境变量：

候选拒绝项被选中时也返回 defer。confidence 与选中项概率不同，适配器分别保留在 metadata 中。HTTP 错误和非法响应明确失败，不自动重试，不将响应正文或认证信息写入工具错误。Jev 只消费显式传入的 JSON，不会读取 state 中的文件路径。

JevSystem1(client=...) 支持注入异步 HTTPX 客户端进行测试或定制传输，注入的客户端由调用者关闭；默认创建的客户端由策略关闭。默认传输不重试、不跟随重定向，并遵循 HTTPX 的代理环境配置；使用 SOCKS 代理需安装 httpx[socks]，自定义客户端的重试策略由调用方控制。

后台脚本沿用 Bash(is_wait=False) 和 ToolControl。Bash 的 timeout 默认仍为 120 秒，应按任务设置。先启动脚本、后启动观测流时，通过任务唯一的配置文件传递 stream call_id/result_dir。文件流默认仅保留最新 20 片，持续消费或自行持久化，不能只在任务结束后假设所有结果仍在。

取消脚本会清理它发起的在途调用，不会自动销毁由 Agent 启动的环境，也不会撤销已经执行的动作。接管前应等待取消完成，再通过环境工具释放控制。系统不保证硬实时，也不会在 Agent Loop 返回后自动因进度文件变化唤醒 System 2。

动态加载与独立 worker

启用管理工具后，模型可以调用 LoadTool，宿主或脚本也可使用相同入口。例如加载上面的计数器：

# runtime 已创建并调用过 enable_management()。
loaded = await runtime.call("LoadTool", arguments={
    "path": "counter_tool.py",
    "factory": "create_tool",
    "config": {},
    "mode": "local",
})
if loaded.is_error:
    raise RuntimeError(loaded.content)

reply = await runtime.call("Counter", method="increment", union_id="counter-a")
print(reply.meta)

path 相对工作目录解析，工厂通过 factory(**config) 调用，可同步或异步返回一个 Tool 实例。同名注册被拒绝；注册成功后，脚本即可调用，模型在下一次请求中看到更新后的工具列表。

mode="local" 在宿主导入并持有实例；mode="process" 将实例放在独立 Python worker 中，宿主注册 RemoteTool 代理。两种模式对 Agent 和脚本提供相同调用接口。加载执行的是本地 Python 代码，工具依赖需要在相应环境中可用。

通信使用仅监听本机 loopback 的 TCP JSON Lines，每次请求使用独立连接，单条消息上限为 8 MiB。worker 流使用专用连接，控制请求使用其他连接；连接故障不会自动重试业务动作，也不会自动重建工具状态。Windows 使用 Job Object 管理子进程树，POSIX 使用独立进程组。

System1 工具支持多个策略实例：tool.register("policy-id", policy) 注册，tool.list() 查询；调用 step 或 steps 时通过 policy_id 选择。省略时优先使用 default，或唯一的已注册实例；存在多个候选时必须明确指定。也可先注册空的 System1Tool()，再调用其 load 方法，以 path、factory、config 和 policy_id 动态创建实例。完整工具的动态导入仍使用 LoadTool，支持本地及独立进程模式。

调用外部命令

有管道、重定向等 shell 语法时使用 Bash；明确程序及参数时使用 Exec，参数按项传递，避免手工拼接引号。优先选择程序提供的 JSON、非交互输出选项。

reply = await runtime.call("Exec", arguments={
    "program": "git", "args": ["status", "--short"]
})
background = await runtime.call("Bash", arguments={
    "command": "python long_job.py"
}, is_wait=False)
# 后续使用 ToolControl 查询状态、输出或取消。

Bash 调用之间不保留 cd 或 export 状态，连续操作放在一次调用或脚本中。Host 与 WSL 分别安装程序和配置身份。此入口不提供 PTY、持续 stdin 或交互终端。

MCP 编程接入

统一发行包默认包含 MCP 依赖，无需额外安装。界面配置方式见上方“界面版 → MCP 配置”。

import asyncio
from pathlib import Path
from loopy import Agent, ToolRuntime
from loopy.mcp import MCPManager, MCPServerConfig

async def main():
    configs = [MCPServerConfig.from_dict({
        "id": "documents", "name": "Documents", "enabled": True,
        "transport": "stdio", "command": "python",
        "args": ["/absolute/path/server.py"],
        "env_refs": {"SERVICE_TOKEN": "DOCUMENT_SERVICE_TOKEN"},
    })]
    async with ToolRuntime(session_id="example", working_dir=Path.cwd()) as runtime:
        await MCPManager(runtime, configs).start()
        agent = Agent(name="assistant", system_prompt="使用工具完成任务",
                      model="claude-sonnet-4-6", working_dir=Path.cwd(),
                      tool_runtime=runtime)
        try:
            await agent.new_session().run("查找项目说明")
        finally:
            await agent.aclose()

asyncio.run(main())

HTTP 定义示例：

{
  "id": "remote-documents",
  "name": "Remote Documents",
  "enabled": true,
  "transport": "streamable_http",
  "url": "https://example.com/mcp",
  "bearer": {"env": "DOCUMENT_SERVICE_TOKEN"},
  "startup_timeout_sec": 30,
  "tool_timeout_sec": 120
}

一个 server 的多个远端工具对应多个 MCPTool 实例，共享该会话内的连接；不同 server 同名工具有不同 Loopy 名称。通过 runtime.list_tools() 或 ToolControl 获取名称及原始工具映射，不手工猜测哈希名称。脚本 ToolClient 使用普通接口：

reply = await client.call(tool_name, method="default",
                          arguments={"query": "example"}, is_wait=True)

模型侧信封为 {"method":"default","arguments":{...},"is_wait":true}。远端的 is_wait、method 等参数保留在 arguments 内，不与 Loopy 控制字段冲突。

生命周期与结果

连接由会话 Runtime 持有，跨模型轮次复用；Agent 关闭不关闭外部 Runtime。Runtime 关闭会取消活动调用并回收连接和子进程。目录变更在下一次准备阶段应用，活动调用继续使用已解析实例；断线后限次重连，不自动重放业务调用。

单次取消不关闭共享连接，也不保证远端副作用撤销。超时、断线后应先检查业务状态，再决定是否重试。服务器返回 isError 时作为普通工具失败进入现有链路。

模型结果保持文本契约。结构化内容提供 JSON 表示；图片、音频、内嵌资源等原始内容保存在调用目录的 mcp_result.json，会话中可下载。不会自动访问 resource_link 或把非文本结果作为模型视觉/音频输入。单个 server 最多 256 个工具、目录描述最多 1 MiB，单次原始结果最多 8 MiB，超限明确失败。

WSL 与开发验证

WSL 需要 Python 3.11+，并在目标发行版安装相同版本的 loopy-agent-fs。MCP 程序、相对目录、环境变量和网络均按 WSL 环境解释；缺少依赖不会切回 Host。测试页可选择 Host / WSL 并指定发行版。

uv run --extra mcp --extra dev python -m pytest tests/mcp
cd ui/backend
uv run --extra dev python -m pytest tests/test_mcp.py

首版不提供 OAuth、旧 HTTP+SSE transport、resources/prompts 交互入口、跨会话共享连接或配置热更新。外部程序可以自行处理认证与网络接入，Loopy 不包含服务商或公司专属登录流程。

上下文压缩与历史回溯

UI 在模型配置中填写“上下文窗口”后，会在每次模型请求前检查预算，并按 Block → Loop → Turn → Session 规划压缩。窗口留空时不启用按水位自动压缩；Agent 仍可使用指定层级的压缩。当前使用保守的 UTF-8 字节估算，界面数值标记为 estimated，并非供应商的精确 token 用量。

SDK 可通过 Agent(..., context_config=ContextConfig(context_window_tokens=200000)) 启用；配置从 loopy.context 导入。恢复已有版本使用 agent.resume_stored_session(session_id)，运行目录、会话 ID 和 Runtime 必须对应原会话。

Agent 通过 Context.inspect 获取稳定引用，通过 Context.compact 排队压缩；既支持指定 block/loop/turn 引用，也支持仅指定层级或 auto。工具结果 queued 仅代表受理，所有工具结果写入后才执行压缩。工具触发和运行时触发都写入同一类压缩标志，下一次模型请求及进程恢复后均能感知历史已压缩。

原始记录保存在会话 context/journal.jsonl，有效版本由 HEAD.json 指向 vNNNNNN.jsonl，父版本和原始记录保留。Agent 可通过 Context.versions/read 按需回溯；UI 对话中显示压缩过程并支持查看历史版本。更多设计和验收范围见 dev_v2.md。

结果与事件

工具结果写入工作目录：

<working_dir>/
  tool_result/
    <call_id>/
      status.json       # 执行状态、结果信息与 meta
      result.txt        # 批量结果
      0000.txt          # 流式片段；按数字序号递增
      ...
    workers/            # 独立 worker 的 stdout / stderr 日志
  sessions/
    <session_id>/
      messages.jsonl    # Agent 会话与模型事件
      tools.jsonl       # Runtime 默认记录的会话级工具事件

流式片段默认保留最近 20 个，文件序号可以超过四位。轮询时按数字序号判断新片段，允许旧片段已被轮转删除。图片等二进制结果由工具写文件，并在文本或结构化内容中返回路径。

脚本接收包含 call_id、status、result_dir、content、is_error 和 meta 的回执。模型端的批量结果超过估算的 4096 tokens 时，返回文件路径和短预览。后台结果不会自动追加到模型消息历史。

事件包括会话创建与恢复、模型请求与响应、工具开始／片段／结束／取消，以及消息记录。通过 Agent 的 event_sinks 观察模型和会话事件。Runtime 默认写入 tools.jsonl 并在自身关闭时关闭日志；传入自定义 sink 后由该 sink 接收工具事件，其持久化和关闭由调用方负责，Agent 关闭不会影响它。UI 在会话级组合工具日志与事件队列。

大结果截断使用 Runtime 的 estimate_tokens 配置，默认按字符数估算；Agent 不覆盖该设置。

HTTP 与事件协议

GET /api/v3/sessions/{id}/turns 支持 limit（1–200，默认 30）和 before（轮次 ordinal）；响应的 next_before 为下一页游标，null 表示没有更早历史。待回答问题通过 GET /api/v2/sessions/{id}/questions 查询。

WebSocket 启用认证时，需在连接后 5 秒内发送首条 JSON 消息 {"type":"authenticate","token":"<backend-token>"}，令牌不放入 URL。升级时需同步更新前端与 Backend。

代码组织、测试与发布

src/loopy/                 # 核心 SDK：Agent、Session、Loop、工具、MCP、Context、System1
src/loopy/application/     # 公共应用层与应用 SDK
src/loopy/entry/           # 统一 loopy 命令入口
src/loopy/cli/             # 终端交互与本地项目操作
src/loopy/gateway/         # 飞书接入、会话路由与消息投递
ui/backend/               # FastAPI 服务与嵌入式桥接
ui/frontend/              # React / Taro 界面
ui/native/                # Electron 宿主与原生接入边界
tests/                    # 核心 SDK 与公共应用层测试
docs/                     # 需求与设计文档

在仓库根目录运行 Python 测试：

uv sync --extra dev --frozen
uv run python -m pytest tests -q
uv sync --project ui/backend --extra dev --frozen
uv run --project ui/backend python -m pytest ui/backend/tests -q

前端在 ui/frontend 执行 npm test、npm run typecheck 和 npm run build:h5。核心 Python 包在仓库根目录执行 uv build。测试使用临时目录与模拟工具、模型，不需要真实模型请求。

CI 在 PR / main 执行检查；只有推送 v* tag 才触发发布，tag 与全部发行包版本须一致。Python 包只构建一次，同一份 wheel / sdist 通过 Windows、macOS、Linux 安装验证，并在所有产品构建通过后，通过 pypi Environment 的 Trusted Publishing 发布 loopy-agent-fs。GitHub Release 仅包含网页静态文件和 Windows / macOS 桌面包，CLI 与飞书 Gateway 随 Python 包安装。PyPI 需预先绑定 gaoxinge/loopy 的 release.yml 和 pypi 环境；签名与公证需另行配置。详细范围见 多端方案。

本项目采用 MIT License。