一小时基于pi-mono实现类似openclaw的agent系统，pi-agent上手实践（云端agent系列二）

上一期我们实现了云端沙盒的搭建，以及简单的调用。这一期我们将使用pi-mono这个agent开发框架/工具链，来实现一个简单的agent系统。

这一期我们将实现以下目标。

新建一个agent服务用来连接pi-agent
调试e2b沙盒，以及agent服务的连接
实现skill系统

TIP
pi-mono是一个基于typescript的agent开发框架/工具链，openclaw能实现快速迭代和开发全依赖这个开源项目的超速更新以及工具提供。pi-agent是其提供的一个已经搞好的agent，我们可以直接使用。

开整#

上一期我们整到了这一步。

e2b-3

接下来我们安装pi-agent。

npm install -g @mariozechner/pi-coding-agent

我们本地安装后，实际上是一个全局命令，相当于本地装了一整个智能体，我们可以在任何地方使用。我们设计的这个系统需要和pi-agent进行通信，以调用pi-agent内部 /指令，以及skill等工具方法。

我们需要简单了解下这个agent框架的组成如下。

包名	npm 包	职责
ai	`@mariozechner/pi-ai`	统一多模型LLM API抽象层
agent	`@mariozechner/pi-agent-core`	Agent运行时（工具调用、事件流、状态管理）
coding-agent	`@mariozechner/pi-coding-agent`	交互式编码智能体CLI与SDK
mom	`@mariozechner/pi-mom`	Slack Bot，将消息委托给编码 Agent
tui	`@mariozechner/pi-tui`	终端UI库（差量渲染）
web-ui	`@mariozechner/pi-web-ui`	AI对话Web组件
pods	`@mariozechner/pi-pods`	GPU Pod上vLLM部署管理CLI

我们要实现的是简单的agent的调度，以及模型层。所以，我们将安装ai/agent/web-ui这三个包。

安装依赖并启动了 pi-agent-service (运行在 4001 端口)。

新建独立的 Agent 服务#

地址：/workspace/pi-agent-service

初始化全新的 Node.js + Express 项目，

项目内安装agent做初始化设置

# 因为我们在远端需要也安装这个agent，所以需要在这个项目中安装
npm install -g @mariozechner/pi-coding-agent

实现 /api/chat 接口：在使用配置的大模型 API Key 初始化 createAgentSession 后，接收用户原始输入并将其交由 pi-agent 进行处理和思考，最终提取出它回答或规划的文本内容。

// index.ts
app.post('/api/chat', async (req, res) => {
  const { input, llmConfig, sandboxId, sandboxApiKey } = req.body;
  if (!input) {
    return res.status(400).json({ error: 'Input is required' });
  }

  // Handle setting API keys for PI runtime
  if (llmConfig) {
    if (llmConfig.type === 'anthropic') {
      process.env.ANTHROPIC_API_KEY = llmConfig.apiKey;
    } else {
      process.env.OPENAI_API_KEY = llmConfig.apiKey;
      if (llmConfig.baseUrl) process.env.OPENAI_BASE_URL = llmConfig.baseUrl;
    }
  }

  // Re-connect to the existing Sandbox
  let sandbox: Sandbox | null = null;
  if (sandboxId && sandboxApiKey) {
    try {
      sandbox = await Sandbox.connect(sandboxId, { apiKey: sandboxApiKey });
      setActiveSandbox(sandbox);
    } catch (e: any) {
      console.warn("Could not connect to sandbox:", e.message);
    }
  }

  try {
    const loader = new DefaultResourceLoader({
      cwd: __dirname,
      agentDir: getAgentDir(),
      // Use the extension factory to dynamically register our 'computer' tool
      extensionFactories: [registerComputerUseExtension]
    });
    await loader.reload();
    
    const { session } = await createAgentSession({
      resourceLoader: loader,
      sessionManager: SessionManager.inMemory()
    });

    // Start processing input
    console.log(`Processing user input through pi-agent: ${input}`);
    
    // Provide initial screenshot to agent if sandbox exists
    let initialPrompt: any = input;
    if (sandbox) {
      try {
        const initialScreenshot = await sandbox.screenshot();
        const initialScreenshotBase64 = Buffer.from(initialScreenshot).toString('base64');
        initialPrompt = [
          { type: 'text', text: input },
          { type: 'image_url', image_url: { url: `data:image/png;base64,${initialScreenshotBase64}` } }
        ];
      } catch (e: any) {
        console.warn("Failed to get initial screenshot:", e.message);
      }
    }

    // Subscribe to event stream and stream back as SSE
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');

    const sendEvent = (event: string, data: any) => {
      res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
    };

    let responseText = '';

    session.subscribe((event) => {
      if (event.type === 'message_update') {
        const msgEvent = event.assistantMessageEvent;
        if (msgEvent.type === 'text_delta') {
          responseText += msgEvent.delta;
          sendEvent('status', { message: `pi-agent 正在思考... (${responseText.length} chars)` });
        } else if (msgEvent.type === 'tool_call') {
          sendEvent('action', { description: `pi-agent 正在调用工具: ${msgEvent.toolCall.function.name}` });
        }
      } else if (event.type === 'tool_result') {
        sendEvent('status', { message: `pi-agent 工具执行完毕: ${event.toolCall.function.name}` });
      }
    });

    try {
      await session.prompt(initialPrompt);
      sendEvent('done', { response: responseText, success: true });
      res.end();
    } catch (err: any) {
      console.error('Session prompt error:', err);
      sendEvent('error', { error: err.message, stack: err.stack });
      res.end();
    }
  } catch (error: any) {
    console.error('Agent setup error:', error);
    res.status(500).json({ error: error.message, stack: error.stack });
  }
});

测试连通性 pi-1

对接 E2B 沙盒流程#

地址：/workspace/api/routes/agent.ts

在用户点击发送指令后，主后端服务会首先调用 http://localhost:4001/api/chat 。将“先过一遍 agent 系统”处理后生成的详细计划或完善后的指令作为 processedInstruction，再真正下发给 E2B 沙盒进行控制。

pi-2

示例代码如下

// agent.ts
import { Router } from 'express'
import { getSandbox } from '../services/e2b.js'
// Removed custom agentLoop from llm.ts since pi-agent handles it now

const router = Router()

router.post('/', async (req, res) => {
  const { sandboxId, instruction, llmConfig, sandboxApiKey } = req.body

  if (!sandboxId || !instruction || !llmConfig || !llmConfig.apiKey) {
    res.status(400).json({ success: false, error: 'Missing sandboxId, instruction, or llmConfig (apiKey required)' })
    return
  }

  const sandbox = getSandbox(sandboxId)
  if (!sandbox) {
    res.status(404).json({ success: false, error: 'Sandbox not found' })
    return
  }

  // Setup SSE
  res.setHeader('Content-Type', 'text/event-stream')
  res.setHeader('Cache-Control', 'no-cache')
  res.setHeader('Connection', 'keep-alive')

  const sendEvent = (event: string, data: any) => {
    res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)
  }

  try {
    sendEvent('status', { message: '正在启动 pi-agent 进行自主屏幕操作...' })
    
    // Call pi-agent-service to process the instruction
    try {
      const piAgentRes = await fetch('http://localhost:4001/api/chat', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ 
          input: instruction, 
          llmConfig,
          sandboxId,
          sandboxApiKey
        })
      });

      if (!piAgentRes.ok) {
        const errData = await piAgentRes.json().catch(() => ({}));
        throw new Error(errData.error || `HTTP error! status: ${piAgentRes.status}`);
      }

      // Stream the response from pi-agent-service to the frontend
      const reader = piAgentRes.body?.getReader();
      if (!reader) throw new Error("No response body reader available");

      const decoder = new TextDecoder('utf-8');
      let done = false;
      let finalResponse = '';

      while (!done) {
        const { value, done: readerDone } = await reader.read();
        if (readerDone) {
          done = true;
          break;
        }

        const chunk = decoder.decode(value, { stream: true });
        const lines = chunk.split('\n');

        for (let i = 0; i < lines.length; i++) {
          const line = lines[i];
          if (line.startsWith('event: ')) {
            const eventName = line.substring(7).trim();
            const dataLine = lines[i + 1];
            if (dataLine && dataLine.startsWith('data: ')) {
              const dataStr = dataLine.substring(6).trim();
              try {
                const dataObj = JSON.parse(dataStr);
                
                if (eventName === 'status') {
                  sendEvent('status', { message: dataObj.message });
                } else if (eventName === 'action') {
                  sendEvent('action', { description: dataObj.description });
                } else if (eventName === 'done') {
                  finalResponse = dataObj.response;
                  sendEvent('status', { message: `pi-agent 任务执行完成: ${finalResponse.substring(0, 50)}...` });
                  sendEvent('done', { message: 'Task completed' });
                  res.end();
                  return; // End the request cleanly
                } else if (eventName === 'error') {
                  console.error("pi-agent runtime error:", dataObj);
                  sendEvent('error', { message: `pi-agent 运行时异常: ${dataObj.error}` });
                  res.end();
                  return;
                }
              } catch (e) {
                console.error("Failed to parse SSE data:", dataStr);
              }
              i++; // skip data line
            }
          }
        }
      }

    } catch (e: any) {
      console.error("Failed to connect to pi-agent-service:", e.message);
      sendEvent('error', { message: `pi-agent 服务调度失败: ${e.message}` });
      res.end();
    }
  } catch (error: any) {
    sendEvent('error', { message: error.message });
    res.end();
  }
})

export default router

前端 UI 增加技能管理#

地址：/workspace/src/pages/Home.tsx

1. 实现 /api/skills 接口（支持 multer 文件上传）#

专门用于接收并保存遵循 pi-agent 规范的 SKILL.md 文件至 .pi/skills/ 目录下

TIP
目前只支持单文件，因为社区有大量方案，所以我只实现基础的，后续在约四期会完善。

// skills.ts
/**
 * 2. POST /api/skills
 * Upload a new skill (SKILL.md file)
 */
app.post('/api/skills', upload.single('skillFile'), (req, res) => {
  if (!req.file) {
    return res.status(400).json({ error: 'Skill file (SKILL.md) is required' });
  }
  
  res.json({
    success: true,
    message: `Skill uploaded successfully to ${req.file.path}`
  });
});

/**
 * 3. GET /api/skills
 * List available uploaded skills
 */
app.get('/api/skills', async (req, res) => {
  try {
    const loader = new DefaultResourceLoader({
      cwd: __dirname,
      agentDir: getAgentDir(),
    });
    await loader.reload();
    
    const { skills } = loader.getSkills();
    res.json({
      success: true,
      skills: skills.map(s => ({ name: s.name, description: s.description }))
    });
  } catch (error: any) {
    res.status(500).json({ error: error.message });
  }
});

2. 前端 UI 增加技能管理模块#

在设置页签底部新增了 "pi-agent 技能管理 (Skills)" 模块。直接点击“上传技能文件 (SKILL.md)”，系统会自动将其存入 pi-agent-service 的技能目录中。界面会自动拉取并展示当前 pi-agent 成功加载的所有扩展技能及其描述。

pi-4

下一步#

本期我们基本上把架子搭起来了，而现在距离真正能使用还有一点距离

下一期将进一步完成以下目标。

实现mcp协议和联通mcp市场
预装软件到e2b沙盒

在大概到四期的时候，我将完善修复bug之后并开源。