零基础教程：230行代码实现OpenClaw

2026-06-20阅读 0热度 0

OpenClaw

1. 连接你和 Agent 的通道：飞书机器人

OpenClaw 要跟飞书消息交互，最直接的桥梁就是飞书机器人。第一步先创建一个机器人。按照飞书开放平台指引，创建应用、创建机器人、授权，完成基础配置。

1.1 为机器人授权

应用创建完成后，在左侧导航栏找到「机器人」→ 点击创建机器人。然后进入「权限管理」，勾选所需权限。那个 JSON 文件可以直接从上方连接复制过来。

1.2 发布机器人

最后别忘记发布应用，确保所有修改都已生效，效果见下图。

现在回到飞书，搜索刚刚创建的机器人，应该能搜到。点进去会发现没有聊天框？别急，这是因为还没配置事件订阅。可以先写代码，等程序跑起来再配置。

2. 让小龙虾张嘴说话：回复消息

正式进入开发环节。先安装一个非常好用且高性能的 Python 包管理器 —— uv。装好后运行 uv --version 检查版本，如果报错，大概率是环境变量没配好，问一下 AI 就能解决。创建一个空文件夹，比如 my_claw，进去之后 uv init 初始化工程（跟 npm init 类似），目录结构如下所示。

那怎么让飞书机器人和本地服务连起来呢？不用自己琢磨，飞书官方已经提供了示例代码，直接复制到 main.py 里就行。

import lark_oapi as lark
from lark_oapi.api.im.v1 import *
import json

# 注册消息接收事件，处理收到的消息。
# Register event handler to handle received messages.
# https://open.feishu.cn/document/uAjLw4CM/ukTMukTMukTM/reference/im-v1/message/events/receive
def do_p2_im_message_receive_v1(data: P2ImMessageReceiveV1) -> None:
    res_content = ""
    if data.event.message.message_type == "text":
        res_content = json.loads(data.event.message.content)["text"]
    else:
        res_content = "解析消息失败，请发送文本消息\nparse message failed, please send text message"
    content = json.dumps({"text": f'收到你发送的消息：{res_content}\nReceived message:{res_content}'})
    if data.event.message.chat_type == "p2p":
        request = (
            CreateMessageRequest.builder()
            .receive_id_type("chat_id")
            .request_body(
                CreateMessageRequestBody.builder()
                .receive_id(data.event.message.chat_id)
                .msg_type("text")
                .content(content)
                .build()
            )
            .build()
        )
        # 用发送 OpenAPI 发送消息
        # Use send OpenAPI to send messages
        # https://open.feishu.cn/document/uAjLw4CM/ukTMukTMukTM/reference/im-v1/message/create
        response = client.im.v1.message.create(request)
        if not response.success():
            raise Exception(f"client.im.v1.message.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}")
    else:
        request: ReplyMessageRequest = (
            ReplyMessageRequest.builder()
            .message_id(data.event.message.message_id)
            .request_body(
                ReplyMessageRequestBody.builder()
                .content(content)
                .msg_type("text")
                .build()
            )
            .build()
        )
        # 用回复 OpenAPI 回复消息
        # Use send OpenAPI to send messages
        # https://open.larkoffice.com/document/server-docs/im-v1/message/reply
        response: ReplyMessageResponse = client.im.v1.message.reply(request)
        if not response.success():
            raise Exception(f"client.im.v1.message.reply failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}")

# 注册事件回调
# Register event handler.
event_handler = (
    lark.EventDispatcherHandler.builder("", "")
    .register_p2_im_message_receive_v1(do_p2_im_message_receive_v1)
    .build()
)

# 创建 LarkClient 对象，用于请求 OpenAPI, 并创建 LarkWSClient 对象，用于通过长连接接收事件。
# Create LarkClient object for requesting OpenAPI, and create LarkWSClient object for receiving events using long connection.
client = lark.Client.builder().app_id(lark.APP_ID).app_secret(lark.APP_SECRET).build()
wsClient = lark.ws.Client(
    lark.APP_ID,
    lark.APP_SECRET,
    event_handler=event_handler,
    log_level=lark.LogLevel.DEBUG,
)

def main():
    #启动长连接，并注册事件处理器。
    #Start long connection and register event handler.
    wsClient.start()

if __name__ == "__main__":
    main()

把代码粘贴进 main.py 后，导入的部分肯定会飘红，因为还没装 lark_oapi。在项目里执行 uv add lark_oapi 即可。调试方便的话，顺便装个 uv add loguru。另外，~~注意力惊人的~~你肯定注意到了，示例代码里没有填写 APP_ID 和 APP_SECRET。去飞书应用主页找到它们复制下来。

但千万别直接把这两个值明文写进代码里——GitHub 上搜 DEEPSEEK_API_KEY=sk- 能搜出一大堆的血淋淋教训摆在那。作为高级专业的开发者，当然要用环境变量。

先在 .gitignore 里加一行 .env，防止不小心推到远程仓库。然后创建 .env 文件，写入 APP_ID 和 APP_SECRET。

怎么把它加载到环境变量呢？Python 有个库 dotenv，专干这事。安装 uv add dotenv，然后在 main.py 中加上：

from dotenv import load_dotenv
import os

load_dotenv() # 把 .env 文件里的变量放到环境变量里
APP_ID = os.getenv("APP_ID") # 获取环境变量，Node.js 里是 process.env.xx
APP_SECRET = os.getenv("APP_SECRET")

# 个人喜欢把创建 client 实例移动到顶部
client = lark.Client.builder().app_id(APP_ID).app_secret(APP_SECRET).build()
# 剩余其他代码...

现在运行一下试试，uv run main.py，如果启动成功，回到飞书应用页面，在「事件与回调」页将「事件配置」和「回调配置」都保存为「长连接」。事件配置需要添加「接收消息」。 ⚠️注意：如果没有启动代码，此处无法选中！

干得不错，发布一下应用：

回到飞书，发现机器人出现了输入框，说明一切正常。

停止刚才的代码，再次 uv run main.py 启动，顺利！

3. 让小龙虾变得聪明：接入 LLM

到现在为止，机器人只能回复固定消息。在 LLM 出现之前，Siri、小爱同学这类语音助手靠文本分类+实体识别+规则判断来决定回复内容。

LLM 诞生后，靠 Next Token Prediction 生成文本，并在 Scaling law 的魔力下催生了“智能”。现在就把 LLM 接进来。本文选的是 DeepSeek 模型——国产模型都挺便宜的，充十块钱能玩很久，用不完还能退，你也可以选别的。先去 Kimi/DeepSeek 后台把 API Key 复制下来，写到 .env 文件里。

openai 库已经封装好了模型调用，安装它：uv add openai。然后在代码中导入并创建实例：

from openai import OpenAI

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY") # 根据实际情况写
llm_client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url="https://api.deepseek.com" # 根据实际情况写
)

然后改造消息接收函数（这里为了简洁，移除了非 p2p 消息的处理）：

def do_p2_im_message_receive_v1(data: P2ImMessageReceiveV1) -> None:
    res_content = ""
    if data.event.message.message_type == "text":
        ###### 新增 ##########
        response = llm_client.chat.completions.create(
            model="deepseek-chat",
            messages=[
                {"role": "system", "content": "You are a helpful assistant"},
                {"role": "user", "content": data.event.message.content},
            ],
            stream=False,
        )
        res_content = response.choices[0].message.content
    else:
        res_content = "解析消息失败，请发送文本消息"

    content = json.dumps({"text": res_content})
    if data.event.message.chat_type == "p2p":
        request = (
            CreateMessageRequest.builder()
            .receive_id_type("chat_id")
            .request_body(
                CreateMessageRequestBody.builder()
                .receive_id(data.event.message.chat_id)
                .msg_type("text")
                .content(content)
                .build()
            )
            .build()
        )
        response = client.im.v1.message.reply(request)
        if not response.success():
            raise Exception(f"client.im.v1.message.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}")

现在我们就得到了一个能用 LLM 回复消息的机器人。不过仔细看一下代码：每次发消息，对于 LLM 来说都是全新的，因为 llm_client.chat.completions.create 里的 messages 参数每次都重新创建。所以得把上下文保存起来——那就用一个全局列表变量吧。

all_messages = [{"role": "system", "content": "You are a helpful assistant"}]

# 添加消息
def add_message(role: Literal["user", "assistant"], content: str):
    all_messages.append({"role": role, "content": content})

def do_p2_im_message_receive_v1(data: P2ImMessageReceiveV1) -> None:
    user_content = data.event.message.content
    res_content = ""
    if data.event.message.message_type == "text":
        add_message(user_content) # 添加消息
        response = llm_client.chat.completions.create(
            model="deepseek-chat",
            messages=all_messages,
            stream=False,
        )
        res_content = response.choices[0].message.content
        add_message("assistant", res_content) # 添加消息
        logger.info(f"LLM response: {res_content}")
    else:
        res_content = "解析消息失败，请发送文本消息"

    content = json.dumps({"text": res_content})
    if data.event.message.chat_type == "p2p":
        request = (
            CreateMessageRequest.builder()
            .receive_id_type("chat_id")
            .request_body(
                CreateMessageRequestBody.builder()
                .receive_id(data.event.message.chat_id)
                .msg_type("text")
                .content(content)
                .build()
            )
            .build()
        )
        response = client.im.v1.message.reply(request)
        if not response.success():
            raise Exception(f"client.im.v1.message.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}")

重新运行测试一下：

效果不错！如果想在程序关闭后依然保留记忆，可以把上下文存到 .jsonl 文件里，或者用 langchain 这类框架。这里为了简洁就不再展开了。

4. 让虾钳拿起武器：准备工具

Claude Code 可以说是 2025 年最火的 coding agent，没有之一。早先的 Copilot 或 Cursor 都是在原有 IDE GUI 基础上做扩展，而 Claude Code 则用 TUI 作为开发者与 Agent 之间的桥梁，重新定义了 AI 辅助软件开发。如果还不了解 Claude Code 的原理，可以参考仓库 learn-claude-code。它的核心就是工具调用 + 循环，这个过程通常被称为 agent loop，如下图所示。

用伪代码可以这样理解：

# 初始化
current_message = lark_chat.user_message
final_reply = None

# 循环直到得到最终回复
while True:
    llm_response = chat.completion(current_message)

    # 如果 LLM 想调用工具，更新消息继续循环
    if llm_response.use_tool:
        current_message = llm_response.message
        continue

    # 如果 LLM 给出最终回复
    if llm_response.final:
        final_reply = llm_response.message
        break

# 结束循环
# 给用户回复
lark_chat.reply(final_reply)

其实只要提供一个命令行工具，LLM 就能完成大量任务：创建文件 touch、查看文件 cat、修改文件 echo、定时任务 crontab 等等。现在也有基于 Claude Code 的 OpenClaw 类似物实现，比如 nanoclaw。巧的是，openai 的 completions 方法有个参数叫 tools，接收一个列表，里面可以放工具名、描述、参数等。用法如下：

response = client.chat.completions.create(
    messages=messages,
    tools=[ # 这里！
        {
            "type": "function",
            "name": "cmd_tool",
            "description": "Run a shell command, e.g. `ls -l`",
            "parameters": {
                # 省略
            }
        }
    ]
)

开始编写这个工具。创建一个 cmd_tool.py，将工具定义为一个类，包含一个静态方法返回 openai 需要的工具定义，还有一个方法执行命令。

import subprocess

class CmdTool:
    def __init__(self):
        pass

    @staticmethod
    def get_definition():
        """OpenAI/DeepSeek 的 API 需要特定的格式：
        {"type": "function","function": {"name": "...","description": "...","parameters": {...}}}
        """
        return {
            "type": "function",
            "function": {
                "name": "run_cmd",
                "description": "Run a shell command, e.g. `ls -l`, `pwd`, `touch test.txt`",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "cmd": {
                            "type": "string",
                            "description": "The shell command to run",
                        },
                    },
                    "required": ["cmd"],
                },
            },
        }

    def execute(self, cmd: str):
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
        return f"stdout: {result.stdout}\nstderr: {result.stderr}"

再创建一个 loop.py 文件，写上 agent loop 逻辑：

import json
from cmd_tool import CmdTool
from openai import Client
from openai.types.chat.chat_completion import ChatCompletion

class AgentLoop:
    def __init__(self, llm_client: Client, all_messages, max_iterations: int = 10):
        self.used_tools = []
        self.llm_client = llm_client
        # all_messages 是列表，引用类型，可以直接在这里修改
        self.all_messages = all_messages
        # 最大迭代次数
        self.max_iterations = max_iterations
        # 本地存在的工具
        self.toolsMap = {"run_cmd": CmdTool()}
        # 工具定义，用于 openai 的 SDK
        self.tools_definitions = [CmdTool.get_definition()]

    # 记录LLM决定的调用
    def add_assistant_message(self, tool_call_id: str, tool_name: str, tool_arguments: dict):
        self.all_messages.append({
            "role": "assistant",
            "content": None,
            "tool_calls": [{
                "id": tool_call_id,
                "type": "function",
                "function": {
                    "name": tool_name,
                    "arguments": json.dumps(tool_arguments),
                },
            }],
        })

    # 记录调用结果
    def add_tool_message(self, tool_call_id: str, result: str):
        self.all_messages.append({
            "role": "tool",
            "content": result,
            "tool_call_id": tool_call_id,
        })

    def add_user_message(self, content: str):
        self.all_messages.append({"role": "user", "content": content})

    def run(self, lark_user_content):
        self.add_user_message(lark_user_content)
        iteration = 0
        while iteration < self.max_iterations:
            iteration += 1
            response: ChatCompletion = self.llm_client.chat.completions.create(
                model="deepseek-chat",
                messages=self.all_messages,
                stream=False,
                # 工具参数是一个列表，可以放多个，此处只放一个
                tools=self.tools_definitions,
                tool_choice="auto",
            )
            tool_calls = response.choices[0].message.tool_calls
            content = response.choices[0].message.content

            if response.choices[0].message.tool_calls and len(tool_calls) > 0:
                for tool_call in tool_calls:
                    tool_name = tool_call.function.name
                    tool_arguments = json.loads(tool_call.function.arguments)
                    # 记录使用过的工具
                    self.used_tools.append(tool_name)
                    # 记录LLM决定的调用
                    self.add_assistant_message(tool_call.id, tool_name, tool_arguments)
                    # 执行工具
                    result = self.toolsMap[tool_name].execute(**tool_arguments)
                    # 记录工具执行结果
                    self.add_tool_message(tool_call.id, result)
            else:
                # 没有工具调用，直接返回
                return content, self.used_tools

然后改造 main.py 中的回复消息函数：

import lark_oapi as lark
from lark_oapi.api.im.v1 import *
import json
from dotenv import load_dotenv
from loguru import logger
import os
from openai import OpenAI
from typing import Literal
from loop import AgentLoop

load_dotenv()
APP_ID = os.getenv("APP_ID")
APP_SECRET = os.getenv("APP_SECRET")
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
logger.info(f"APP_ID: {APP_ID}")
logger.info(f"APP_SECRET: {APP_SECRET}")

llm_client = OpenAI(api_key=DEEPSEEK_API_KEY, base_url="https://api.deepseek.com")
all_messages = [{"role": "system", "content": "You are a helpful assistant"}]
agent_loop = AgentLoop(llm_client, all_messages)

lark_messages_set = set()

# 注册接收消息事件，处理接收到的消息。
# Register event handler to handle received messages.
# https://open.feishu.cn/document/uAjLw4CM/ukTMukTMukTM/reference/im-v1/message/events/receive
def do_p2_im_message_receive_v1(data: P2ImMessageReceiveV1) -> None:
    user_content = data.event.message.content
    res_content = ""
    if data.event.message.message_id in lark_messages_set:
        logger.info(f"重复消息，message_id: {data.event.message.message_id}")
        return
    else:
        lark_messages_set.add(data.event.message.message_id)

    if data.event.message.message_type == "text":
        res_content, used_tools = agent_loop.run(user_content)
    else:
        res_content = "解析消息失败，请发送文本消息"

    content = json.dumps({"text": res_content + f" 已使用工具：{', '.join(used_tools)}"})
    if data.event.message.chat_type == "p2p":
        request = (
            CreateMessageRequest.builder()
            .receive_id_type("chat_id")
            .request_body(
                CreateMessageRequestBody.builder()
                .receive_id(data.event.message.chat_id)
                .msg_type("text")
                .content(content)
                .build()
            )
            .build()
        )
        response = client.im.v1.message.reply(request)
        if not response.success():
            raise Exception(f"client.im.v1.message.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}")

# 注册事件回调
# Register event handler.
event_handler = (
    lark.EventDispatcherHandler.builder("", "")
    .register_p2_im_message_receive_v1(do_p2_im_message_receive_v1)
    .build()
)

# 创建 LarkClient 对象，用于请求OpenAPI, 并创建 LarkWSClient 对象，用于使用长连接接收事件。
# Create LarkClient object for requesting OpenAPI, and create LarkWSClient object for receiving events using long connection.
client = lark.Client.builder().app_id(APP_ID).app_secret(APP_SECRET).build()
wsClient = lark.ws.Client(
    APP_ID,
    APP_SECRET,
    event_handler=event_handler,
    log_level=lark.LogLevel.DEBUG,
)

def main():
    #启动长连接，并注册事件处理器。
    #Start long connection and register event handler.
    wsClient.start()

if __name__ == "__main__":
    main()

测试一下效果，问两个问题：

大功告成！但有时会遇到飞书消息重复推送的情况。

按官方文档处理一下，设置一个集合来去重：

# 省略代码...
lark_messages_set = set()

def do_p2_im_message_receive_v1(data: P2ImMessageReceiveV1) -> None:
    # 省略其他代码...
    if data.event.message.message_id in lark_messages_set:
        logger.info(f"重复消息，message_id: {data.event.message.message_id}")
        return
    else:
        lark_messages_set.add(data.event.message.message_id)
    # 省略其他代码...

其实定时任务也是一个工具，可以参考 nanobot 的实现。加载 SKILLS 也不难，就是把它的说明文件加载到 system prompt 中，例如：

parts.append(f"""# Skills
The following skills extend your capabilities. To use a skill, read its SKILL.md file using the read_file tool.
Skills with available="false" need dependencies installed first - you can try installing them with apt/brew.
{skills_summary}
""")

这里就不展开讲了。现在，我们已经实现了一个能调用工具的 Agent，完结撒花 ????