环境信息

• Python 3.12+
• Anthropic Python SDK：按当前官方 SDK 安装即可；示例通过ANTHROPIC_MODEL环境变量指定模型，不把易变的模型名写死在业务逻辑里。
• 香港天文台当前天气 JSON 接口：rhrread，数据由香港天文台提供，官方数据平台标注为“每小时及有更新时”更新。
• 运行模式：--dry-run不需要API Key；真实调用需要配置ANTHROPIC_API_KEY。

我在做香港职场自动化时遇到过一个很像“模型犯错”的问题：Agent查到了天气，也写出了双语简报，但流程没有稳定地进入下一步。有时它停在第一项，有时审核任务重复创建，还有一次接口返回异常，模型却继续往下写，最后生成了一段看起来很完整、实际上没有数据依据的文字。

后来我把问题从“提示词怎么写”改成了“流程状态怎么定义”：读取天气是READ_ONLY，生成草稿是纯函数，创建审核任务是WRITE_CONFIRM。只要写操作没有得到确认，程序就不能向下走。

这篇文章用香港天文台公开数据做一个小型职场简报流程。重点不是让Claude替人发消息，而是让它在该停的地方停下来。

1. 先区分“模型回合”和“业务状态”

Anthropic Messages API返回stop_reason="tool_use"时，说明Claude给出了一个或多个tool_use块，应用程序需要执行工具，再把对应的tool_result放回下一轮消息。stop_reason="end_turn"只表示这一轮自然结束，并不自动等于业务流程的每个步骤都完成。

因此，代码里至少要有这几个状态：

状态	允许做什么	不能做什么
PLANNING	接收任务、准备工具契约	不能执行副作用操作
READ_ONLY	查询公开天气、读取日历等	不能创建任务、发消息
VALIDATE	检查字段、时间戳、结果是否完整	不能把缺失字段当默认值
WAITING_CONFIRM	展示将要执行的写操作	不能自动放行
DONE / FAILED	返回结果或明确错误	不能假装流程成功

这和“写一个while循环”不是同一个层次。循环只是传输消息；状态机负责回答“当前能不能执行这类工具”。

2. 香港公开天气数据：先验证输入，再交给Agent

当前天气接口地址如下：

https://data.weather.gov.hk/weatherAPI/opendata/weather.php?dataType=rhrread&lang=sc

接口响应包含updateTime、temperature、rainfall和预警相关字段。脚本只提取简报需要的字段，避免把整份原始JSON无差别塞入模型上下文。验证时我读取到的快照为：更新时间2026-08-25T11:02:00+08:00，香港天文台参考温度31°C，接口返回的地区最高一小时雨量为4mm。这个数字只是当次读取结果，不是固定天气结论。

数据层有三个细节值得保留：

1. updateTime和观测时间不是一回事，简报里应同时保留来源时间。
2. rainfall.data可能没有min字段，不能直接解包固定结构。
3. 天气数据是观测快照，不是对未来通勤的承诺，文案里不能擅自推断“肯定不会下雨”。

3. 工具契约要写清副作用

我把三个工具拆成两类：get_hko_weather和draft_bilingual_brief是无副作用工具，create_review_task是写工具。后者即使只是“创建一个待审核任务”，也可能产生真实记录，所以默认阻断。

Anthropic官方文档支持在自定义工具中使用strict: true，让工具调用更严格地匹配输入结构。但这只解决参数形状问题，不会替你判断业务权限；“字段合法”和“这一步允许执行”必须分开检查。

4. 完整Python代码：状态机加写操作闸门

下面的代码包含真实天气读取、双语草稿生成、幂等键、最大步数和本地干跑模式。干跑模式用固定的模型回合验证流程：第三步会请求创建审核任务，但没有--confirm时只返回WAITING_CONFIRM，不会执行写工具。

from __future__ import annotations

import hashlib
import json
import os
import sys
import urllib.request
from dataclasses import dataclass
from enum import Enum
from typing import Any


TOOLS = [
    {
        "name": "get_hko_weather",
        "description": "Read Hong Kong Observatory current weather data. No side effect.",
        "strict": True,
        "input_schema": {
            "type": "object",
            "properties": {"lang": {"type": "string", "enum": ["sc", "tc", "en"]}},
            "required": ["lang"],
        },
    },
    {
        "name": "draft_bilingual_brief",
        "description": "Turn a verified weather snapshot into a Chinese-English draft. No side effect.",
        "strict": True,
        "input_schema": {
            "type": "object",
            "properties": {"weather": {"type": "object"}, "audience": {"type": "string"}},
            "required": ["weather", "audience"],
        },
    },
    {
        "name": "create_review_task",
        "description": "Create a review task for a human to inspect a draft. This is a write operation.",
        "strict": True,
        "input_schema": {
            "type": "object",
            "properties": {"title": {"type": "string"}, "draft": {"type": "string"}},
            "required": ["title", "draft"],
        },
    },
]

WRITE_TOOLS = {"create_review_task"}


class State(str, Enum):
    PLANNING = "PLANNING"
    READ_ONLY = "READ_ONLY"
    VALIDATE = "VALIDATE"
    WAITING_CONFIRM = "WAITING_CONFIRM"
    DONE = "DONE"
    FAILED = "FAILED"


@dataclass
class RunResult:
    state: State
    text: str = ""
    pending_tools: tuple[str, ...] = ()
    steps: int = 0


def fetch_hko_weather(lang: str) -> dict[str, Any]:
    url = (
        "https://data.weather.gov.hk/weatherAPI/opendata/"
        f"weather.php?dataType=rhrread&lang={lang}"
    )
    with urllib.request.urlopen(url, timeout=15) as response:
        payload = json.load(response)

    temperatures = payload.get("temperature", {}).get("data", [])
    rainfall = payload.get("rainfall", {}).get("data", [])
    hko = next(
        (row for row in temperatures if row.get("place") == "香港天文台"),
        temperatures[0] if temperatures else {},
    )
    max_rain = max((float(row.get("max", 0)) for row in rainfall), default=0.0)
    return {
        "source": "Hong Kong Observatory",
        "update_time": payload.get("updateTime"),
        "record_time": payload.get("temperature", {}).get("recordTime"),
        "temperature_c": hko.get("value"),
        "rainfall_max_mm": max_rain,
        "warning": payload.get("warningMessage") or payload.get("tcmessage") or "",
    }


def draft_bilingual_brief(weather: dict[str, Any], audience: str) -> dict[str, str]:
    if not weather.get("update_time"):
        raise ValueError("weather snapshot has no update_time")
    rain = weather.get("rainfall_max_mm", 0)
    cn = (
        f"早安，香港天文台数据更新时间为 {weather['update_time']}，"
        f"参考温度 {weather.get('temperature_c')}°C，过去一小时最高雨量 {rain} mm。"
    )
    en = (
        f"Good morning. HKO data updated at {weather['update_time']}; "
        f"reference temperature {weather.get('temperature_c')}°C and "
        f"maximum one-hour rainfall {rain} mm."
    )
    return {"audience": audience, "subject": "Hong Kong morning brief", "draft": cn + "\n" + en}


def idempotency_key(name: str, arguments: dict[str, Any]) -> str:
    raw = json.dumps({"name": name, "arguments": arguments}, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]


class AgentRunner:
    def __init__(self, client: Any, model: str, allow_writes: bool = False, max_steps: int = 6):
        self.client = client
        self.model = model
        self.allow_writes = allow_writes
        self.max_steps = max_steps

    def execute(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]:
        key = idempotency_key(name, arguments)
        if name == "get_hko_weather":
            return {"ok": True, "key": key, "data": fetch_hko_weather(arguments["lang"])}
        if name == "draft_bilingual_brief":
            return {"ok": True, "key": key, "data": draft_bilingual_brief(arguments["weather"], arguments["audience"])}
        if name == "create_review_task":
            if not self.allow_writes:
                raise PermissionError("write tool requires human confirmation")
            return {"ok": True, "key": key, "data": {"status": "accepted", "title": arguments["title"]}}
        raise KeyError(f"unknown tool: {name}")

    def run(self, query: str) -> RunResult:
        messages: list[dict[str, Any]] = [{"role": "user", "content": query}]
        for step in range(1, self.max_steps + 1):
            response = self.client.messages.create(
                model=self.model, max_tokens=1200, tools=TOOLS, messages=messages
            )
            if response.stop_reason == "end_turn":
                text = "\n".join(block.text for block in response.content if block.type == "text")
                return RunResult(State.DONE, text=text, steps=step)
            if response.stop_reason != "tool_use":
                return RunResult(State.FAILED, text=f"unexpected stop_reason={response.stop_reason}", steps=step)

            tool_blocks = [block for block in response.content if block.type == "tool_use"]
            if not tool_blocks:
                return RunResult(State.FAILED, text="tool_use response has no tool block", steps=step)
            pending = tuple(block.name for block in tool_blocks if block.name in WRITE_TOOLS)
            if pending and not self.allow_writes:
                return RunResult(State.WAITING_CONFIRM, pending_tools=pending, steps=step)

            messages.append({"role": "assistant", "content": response.content})
            results = []
            for block in tool_blocks:
                try:
                    result = self.execute(block.name, block.input)
                except Exception as exc:
                    result = {"ok": False, "error": str(exc), "key": idempotency_key(block.name, block.input)}
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(result, ensure_ascii=False),
                })
            messages.append({"role": "user", "content": results})
        return RunResult(State.FAILED, text="max_steps reached", steps=self.max_steps)


class FakeBlock:
    def __init__(self, block_type: str, **kwargs: Any):
        self.type = block_type
        self.__dict__.update(kwargs)


class FakeResponse:
    def __init__(self, stop_reason: str, content: list[FakeBlock]):
        self.stop_reason = stop_reason
        self.content = content


class FakeMessages:
    def __init__(self):
        self.turn = 0

    def create(self, **_: Any) -> FakeResponse:
        self.turn += 1
        if self.turn == 1:
            return FakeResponse("tool_use", [FakeBlock("tool_use", id="w1", name="get_hko_weather", input={"lang": "sc"})])
        if self.turn == 2:
            return FakeResponse("tool_use", [FakeBlock("tool_use", id="w2", name="draft_bilingual_brief", input={"weather": {"update_time": "2026-08-25T11:02:00+08:00", "temperature_c": 31, "rainfall_max_mm": 4}, "audience": "Hong Kong engineering team"})])
        if self.turn == 3:
            return FakeResponse("tool_use", [FakeBlock("tool_use", id="w3", name="create_review_task", input={"title": "Review morning brief", "draft": "draft text"})])
        return FakeResponse("end_turn", [FakeBlock("text", text="The draft is ready and the review task was accepted.")])


def build_fake_client() -> Any:
    class FakeClient:
        def __init__(self):
            self.messages = FakeMessages()
    return FakeClient()


def main() -> None:
    # --dry-run 使用固定回合验证状态机；真实调用时换成 anthropic.Anthropic()。
    if "--dry-run" not in sys.argv:
        from anthropic import Anthropic
        if not os.getenv("ANTHROPIC_MODEL"):
            raise SystemExit("set ANTHROPIC_MODEL before a real API call")
        client = Anthropic()
    else:
        client = build_fake_client()  # 干跑只用于验证状态机，生产环境不使用FakeClient
    result = AgentRunner(
        client,
        model=os.getenv("ANTHROPIC_MODEL", "dry-run-model" if "--dry-run" in sys.argv else ""),
        allow_writes="--confirm" in sys.argv,
    ).run("Create a Hong Kong morning weather brief and send it for human review.")
    print(json.dumps({"state": result.state, "pending_tools": result.pending_tools, "steps": result.steps, "text": result.text}, ensure_ascii=False))

代码里有一个看似保守、实际很重要的判断：发现写工具时，在未确认状态下直接返回，而不是先执行再补一条“请确认”。因为任务创建已经是副作用，确认应该发生在副作用之前。

真实运行时，Anthropic SDK会把响应里的tool_use块原样放入assistant消息，再把每个工具的结果以tool_result放进下一条user消息。一个响应可能包含多个工具调用，所以不能只取第一个块；若要禁止并行调用，可以显式设置tool_choice，但本文的读工具执行器已经按“无副作用、可重复”设计。

5. 干跑验证：故意让流程停在写操作之前

代码中的FakeClient保留了一个固定回合：天气查询、双语草稿、创建审核任务、最终回复四步。执行：

python3 claude_agent_state_machine_demo.py --dry-run

输出应类似：

{"state": "WAITING_CONFIRM", "pending_tools": ["create_review_task"], "steps": 3, "text": ""}

加上确认标记后：

python3 claude_agent_state_machine_demo.py --dry-run --confirm

才会进入DONE。这里的--confirm只是本地演示开关；接入企业系统时，应把它替换成真实的登录用户确认、审批记录或一次性授权，不要把环境变量当成完整权限系统。

6. 我没有把“失败重试”写成无限循环

工具调用的异常必须作为结构化数据回传给Claude，例如：

{
  "ok": false,
  "error": "HTTP 503",
  "key": "f9e1a5c2d0a31a76"
}

这样模型知道这一步失败了，程序也能用key判断是否是同一个请求。生产环境还需要把幂等键落到持久化存储，设置重试次数和退避时间；当前示例只展示边界，不会伪装成完整任务系统。

另外，以下情况不能简单重试：

• 参数结构不合法：先修输入，不要重复发送同一错误。
• 写工具已经返回成功但客户端超时：先按幂等键查询结果，再决定是否补偿。
• 数据缺少更新时间：进入FAILED或人工检查，不能把当前时间填进去。
• Claude返回max_tokens：这是响应被截断，不等于工具成功；需要单独处理。

7. 这篇文章真正解决的“第一步陷阱”

Agent只执行第一步，通常不是模型不会做第二步，而是应用层没有给第二步一个明确的状态入口；Agent重复创建任务，也不一定是模型失控，而可能是工具没有幂等键；Agent在异常数据上继续写文案，则是结果契约没有强制要求update_time。

把流程拆成状态、把工具拆成权限、把结果拆成可验证字段之后，模型只是其中一个决策组件。它可以建议下一步，但不能绕过程序定义的边界。

总结

今天这篇没有再重复“Claude Agent就是一个while循环”的基础解释，而是把重点放在更容易被忽略的工程层：

1. 用stop_reason区分模型自然结束和请求工具。
2. 用状态机区分只读、校验、等待确认和失败。
3. 用strict: true约束工具参数，再用业务代码判断权限。
4. 用真实香港天文台公开数据验证输入，不用固定假数据冒充线上结果。
5. 用幂等键、最大步数和结构化错误，避免重试把问题放大。

如果要继续扩展，我会先接入只读的日历和公司知识库，再考虑创建审核任务；不会一开始就给Agent发送邮件、修改日程或写入生产系统的权限。

参考资料

1. Anthropic Tool use with Claude
2. Anthropic Stop reasons and fallback
3. Hong Kong Observatory Open Data

数据说明：本文使用香港天文台公开当前天气接口做技术验证。天气数值会随接口更新时间变化；示例不构成天气预报，也不替代任何职场或出行判断。创建审核任务部分仅用于演示写操作闸门，未接入真实企业系统。