这一系列我会用 Python一步步手搓一个“可运行、可扩展、可解释”的 OpenClaw 简化版。
第一篇先不追求功能多，而是先搭好最重要的骨架：服务入口、会话并发、最小 AgentLoop。

0. 为什么要手搓 OpenClaw

OpenClaw 很强，但完整工程体量也很大。对于大多数开发者来说，直接阅读全量代码会有三个痛点：

• 模块多：Gateway、Agent、Tools、Sessions、Channels 互相耦合
• 路径长：一条消息从输入到回复，跨越多个子系统
• 调试难：没有自己的“最小版本”，很难定位问题

所以这个系列采用一个更实用的学习路径：
先做最小闭环，再逐步补齐能力。

---

1. 目标

用 Python 从 0 到 1 复现 OpenClaw 的核心能力：

• Agent Loop（工具调用 + 多轮推理）
• Session 与并发隔离
• 记忆系统（短期 + 长期）
• Skills 系统（分层加载）
• Web/Telegram 等渠道接入

第一篇的阶段目标是：

• 跑起 FastAPI 服务
• 打通一个最小 /v1/chat 对话接口
• 具备会话隔离与并发控制（每会话锁 + 全局信号量）

---

2. 目标框架

---

3. 技术选型

• 语言：Python 3.11+
• 服务框架：FastAPI（Web/Dashboard/API 一体）
• 模型接入：先用简化 Provider 抽象，后续切真实 OpenAI Compatible
• 并发模型：asyncio + 每会话 Lock + 全局 Semaphore
• 配置与数据模型：Pydantic v2 + pydantic-settings
• 日志：标准 logging（后续可升级 JSON 结构化日志）
• 任务调度：APScheduler（后续阶段引入）
• 数据库：第一篇先不强依赖，阶段二引入 PostgreSQL + pgvector

---

4. 第一篇完成了什么

本篇配套工程目录：openclaw_py 下载地址

已经有这些关键模块：

• app/main.py：FastAPI 启动入口
• app/api/routes_chat.py：最小聊天接口
• app/session/manager.py：会话复用 + 并发控制
• app/core/agent.py：最小 Agent Loop（先支持 echo 与工具占位）
• app/core/tools.py：工具注册与调用骨架
• run.py：一键启动 uvicorn
• .env 与 requirements.txt：本地运行基础配置

---

5. 如何运行（可复制）

cd openclaw_py
python -m venv .venv
.venv\Scripts\activate
pip install -r requirements.txt
python run.py

服务启动后：

• 健康检查：http://127.0.0.1:7788/health
• OpenAPI 文档：http://127.0.0.1:7788/docs

测试聊天接口（PowerShell）：

curl -X POST "http://127.0.0.1:7788/v1/chat" ^
  -H "Content-Type: application/json" ^
  -d "{\"session_id\":\"demo\",\"message\":\"hello\"}"

---

6. 设计细节：为什么 SessionManager 要先做

很多人一开始只关心 LLM 调用，但真实系统更容易先死在并发上。
如果同一个 session_id 同时进两条消息，不加锁就会出现：

• 历史消息交错写入
• 上下文错乱，回复“串台”
• 工具调用顺序被打乱

所以第一篇就把这一层打好：

• 同一会话：串行（asyncio.Lock）
• 全局会话：限流（asyncio.Semaphore）

这一步是后面做“记忆系统”和“多渠道接入”的地基。

from __future__ import annotations

import asyncio
from contextlib import asynccontextmanager
from typing import AsyncIterator

from app.config import settings
from app.core.agent import Agent


class SessionManager:
    def __init__(self, max_concurrent: int = 4) -> None:
        self._sessions: dict[str, Agent] = {}
        self._locks: dict[str, asyncio.Lock] = {}
        self._semaphore = asyncio.Semaphore(max_concurrent)

    def get_or_create(self, session_id: str) -> Agent:
        if session_id not in self._sessions:
            self._sessions[session_id] = Agent()
        return self._sessions[session_id]

    def get_lock(self, session_id: str) -> asyncio.Lock:
        if session_id not in self._locks:
            self._locks[session_id] = asyncio.Lock()
        return self._locks[session_id]

    @asynccontextmanager
    async def acquire(self, session_id: str) -> AsyncIterator[None]:
        lock = self.get_lock(session_id)
        async with lock:
            async with self._semaphore:
                yield

    def __len__(self) -> int:
        return len(self._sessions)


_session_manager = SessionManager(max_concurrent=settings.max_concurrent_agents)


def get_session_manager() -> SessionManager:
    return _session_manager

---

7. 这篇没做什么

为了保证节奏清晰，第一篇有意不做：

• 不接真实大模型 API（先保留 Provider 接口）
• 不做复杂工具协议（先保留 Tool Runtime 骨架）
• 不做数据库与向量检索（留给阶段二）
• 不做 Telegram/Discord 真接入（留给阶段三）

---

8. 常见坑与排查

• 端口占用：把 .env 里的 PORT 改掉
• 全局环境依赖冲突：优先用 .venv，不要直接污染系统 Python

---

9. 下一篇预告

下一篇我会接入真实的 OpenAI-Compatible Provider，把当前的 EchoProvider 替换掉，内容包括：

• .env 增加 OPENAI_API_KEY 与 OPENAI_BASE_URL
• Provider 抽象如何兼容不同模型厂商
• 超时、重试、错误映射的最小实现

到那时，这个项目就会从“骨架”进化成“真能聊天的 Agent”。

10. 看到这里，不妨支持一下

如果你已经看到这里，说明你对“手搓 OpenClaw”是真的感兴趣。
这套系列会持续把代码和踩坑都开源出来，不走玄学，尽量做到每篇都能复现。

如果这篇对你有帮助，欢迎随手支持一下：

• 点个赞，让我知道这条路线是有价值的
• 点个关注，后续 3~12 篇更新不会错过
• 点个收藏，后面实操时可以随时回来看代码片段
• 有余力的话，来个打赏，我会把更多时间投入到高质量连载里

你的每一次反馈，都会直接影响这个系列更新的速度和深度。 我们下一篇见。