这次我们来看一个非常实际的 AI Agent 应用场景用 Claude Code 从零搭建一个农业物联网监测平台。Claude Code 是 Anthropic 推出的终端 AI 编程代理安装后可以直接在当前项目目录里读写代码、执行命令、跑测试、修 bug。它本身不在本地运行大模型而是通过网络调用云端模型能力所以本机不需要独立显卡也不用下载几十 GB 的模型权重。你只需要一个能跑 Node.js 的终端环境加上有可用权限的账号授权就能开始干活。这篇文章是“Claude Code 100 个案例”系列中的农业物联网案例。目标不是堆概念而是把一条完整链路跑通传感器数据采集 → 数据存储 → REST API → 可视化看板 → 阈值告警 → 历史数据批量回填。整套平台用 FastAPI SQLite 起步单机可跑后续可以平滑换到 PostgreSQL / InfluxDB 这类更强的时间序列存储。读完你会得到几样东西Claude Code 的安装方式和交互用法用提示词让 Claude Code 生成项目骨架的方法一套可以直接复制运行的农业物联网监测平台代码以及接口测试、资源观察和常见问题排查思路。如果你正在做农业信息化、温室大棚监测、果园环境记录或者单纯想看看 Claude Code 这类 AI 编程工具到底能把一个完整项目推进到哪一步这篇文章可以直接收藏。1. 核心能力速览先给结论方便你快速判断这个案例是否值得跟着做。能力项说明项目类型AI 辅助编程 物联网应用开发案例主角工具Claude CodeAnthropic 终端 AI 编程代理平台架构传感器采集 → SQLite 存储 → FastAPI 接口 → Web 看板 → 告警通知主要功能项目脚手架生成、传感器数据上报、时序查询、可视化看板、阈值告警、CSV 批量导入硬件要求普通办公电脑即可建议 8GB 内存以上不需要独立 GPU运行环境Windows / macOS / Linux需要 Node.js 和 Python 3.10Claude Code 启动方式终端交互模式claude非交互模式claude -p平台服务启动方式uvicorn app.main:app启动 Web 服务是否支持 API支持平台提供 RESTful 接口是否支持批量任务支持含定时采集、CSV 历史数据回填数据存储起步用 SQLite生产环境可迁移到 PostgreSQL / InfluxDB适合读者物联网开发者、农业信息化从业者、AI 辅助编程入门者注意一点本文给出的代码是农业物联网平台的常见实现结构属于案例示意。实际让 Claude Code 生成时具体文件名、接口路径、字段命名都会有差异以你本机生成并通过测试的结果为准。2. 适用场景与使用边界这个案例适合谁一句话需要快速验证物联网数据链路的人。不管是温室大棚的空气温湿度、土壤湿度、光照强度还是果园的二氧化碳浓度、气象数据底层的“采集-存储-查询-展示-告警”逻辑是一致的。用 Claude Code 搭建原型能把原来两三天的框架搭建工作压缩到一个下午。但也要说清楚边界。这套平台的定位是原型和中小规模监测不是直接拿去种菜就完事的工业系统。真实生产环境里传感器需要校验、去重、断线续传网关要考虑离线缓存数据库要考虑写入吞吐接口要考虑鉴权、限流、HTTPS。这些在案例里只做了最简版本生产化之前还需要大量加固。另一个必须强调的是数据合规和隐私边界。农业物联网数据可能涉及地块位置、农事记录、产量信息属于经营数据。采集和存储前要确认数据来源合法、相关方已授权不要把未经脱敏的数据上传到未授权的第三方服务如果平台要部署到公网必须给接口加访问控制和身份校验避免传感器数据被任意读取。3. 环境准备与前置条件先列环境清单再逐个确认。3.1 本机工具工具版本建议用途Node.js18 以上安装 Claude Codenpm随 Node.js 自带包管理器Python3.10 以上运行 FastAPI 平台服务Git有即可代码版本管理浏览器任意现代浏览器访问可视化看板确认命令node --version npm --version python --version如果输出正常环境就算准备好了。3.2 安装 Claude CodeClaude Code 的安装很简单用 npm 全局安装npm install -g anthropic-ai/claude-code安装完成后检查版本claude --version第一次运行claude会进入账号授权流程需要 Claude Code 的使用权限和有效登录凭证。如果在受管环境中使用注意企业策略可能限制订阅访问遇到授权类提示时优先找管理员确认账号是否有对应权限。3.3 安装 Python 依赖平台服务部分使用 FastAPI进入项目目录后创建虚拟环境并安装依赖mkdir agri-iot cd agri-iot python -m venv .venv # Windows PowerShell 执行 # .venv\Scripts\activate # macOS / Linux 执行 source .venv/bin/activaterequirements.txt内容如下fastapi0.110 uvicorn[standard]0.29 pydantic2.6 requests2.31 apscheduler3.10 pytest8.0安装pip install -r requirements.txtClaude Code 的核心功能是写代码、改代码、执行命令不是本地推理所以这里没有 CUDA、PyTorch、显卡驱动这些前置依赖。资源瓶颈在 Web 服务和数据库上普通电脑完全带得动。4. 用 Claude Code 设计项目架构装好环境后先不急着写代码让 Claude Code 帮我们把架构理清楚。进入项目目录启动交互模式claude在交互会话里输入请作为资深 Python 工程师帮我设计一个农业物联网监测平台的项目结构。 要求使用 FastAPI SQLite包含传感器数据上报、时序查询、告警记录、 Web 可视化看板、CSV 批量导入。目录结构要清晰模块职责要单一。 先给出整体架构说明再列目录树。Claude Code 会先输出一段架构说明然后给出类似下面的目录结构。不同版本生成结果会有差异但整体思路应该一致agri-iot/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口与路由注册 │ ├── database.py # SQLite 连接与建表 │ ├── models.py # 数据库表结构定义 │ ├── schemas.py # Pydantic 请求/响应模型 │ └── routers/ │ ├── __init__.py │ ├── sensors.py # 传感器数据接口 │ └── alarms.py # 告警记录接口 ├── collector/ │ ├── simulator.py # 模拟传感器 │ └── uploader.py # 数据上报客户端 ├── dashboard/ │ └── index.html # 可视化看板 ├── scripts/ │ ├── init_db.py # 初始化数据库 │ └── batch_import.py # CSV 批量导入 ├── tests/ │ └── test_api.py # 接口测试 ├── requirements.txt └── README.md这个设计的核心思路是分层collector负责采集设备侧数据app负责平台服务dashboard负责展示scripts负责运维类任务。每层之间只通过 HTTP 接口或数据库交互互不耦合。如果你对目录不满意可以直接在会话里追加要求例如把 dashboard 改为用 FastAPI 直接返回单页 HTML不要单独起一个前端服务。Claude Code 会基于对话上下文调整文件结构这是它比普通代码补全工具强的地方。5. 环境初始化与数据库建表架构确定后让 Claude Code 生成数据库模块。这里提供一个可直接使用的app/database.py作用是连接 SQLite 并建表import sqlite3 from pathlib import Path DB_PATH Path(__file__).parent.parent / data / agri_iot.db def get_connection(): DB_PATH.parent.mkdir(parentsTrue, exist_okTrue) conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn def init_db(): with get_connection() as conn: conn.executescript( CREATE TABLE IF NOT EXISTS devices ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, location TEXT, created_at TEXT DEFAULT (datetime(now)) ); CREATE TABLE IF NOT EXISTS sensor_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id INTEGER NOT NULL, ts TEXT NOT NULL, air_temp REAL, air_humidity REAL, soil_humidity REAL, light_lux REAL, co2_ppm REAL, FOREIGN KEY (device_id) REFERENCES devices(id) ); CREATE TABLE IF NOT EXISTS alarm_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id INTEGER, type TEXT, message TEXT, created_at TEXT DEFAULT (datetime(now)) ); )三张表的职责很清楚devices维护设备档案比如“三号大棚”“东区果园”。sensor_data是核心时序数据表每条记录对应一次传感器上报时间字段ts用来做趋势查询。alarm_records保存告警历史方便回看某个时段发生了多少次温度越界。这里用 SQLite 是为了让案例开箱即跑。数据结构稳定后如果要上生产可以把这个建表语句平滑迁移到 PostgreSQL 或 InfluxDB。app/schemas.py定义接口的数据结构from pydantic import BaseModel from typing import Optional class SensorDataIn(BaseModel): device_id: int ts: Optional[str] None air_temp: Optional[float] None air_humidity: Optional[float] None soil_humidity: Optional[float] None light_lux: Optional[float] None co2_ppm: Optional[float] None字段全部设为可选是为了兼容不同传感器上报的字段差异。有的设备只有温湿度有的设备有光照和二氧化碳接收端不需要为每种设备单独写接口。6. 传感器数据采集与设备接入数据接入是物联网平台的第一步。真实场景中传感器通过 Modbus、RS485、LoRa、4G 网关等协议接入案例里我们先用一个模拟器代替物理设备验证整条链条是否通畅。6.1 模拟传感器collector/simulator.py负责每隔几秒生成一条环境数据并上报到平台接口import random import time import requests def read_sensor(): return { device_id: 1, air_temp: round(random.uniform(15, 35), 1), air_humidity: round(random.uniform(40, 80), 1), soil_humidity: round(random.uniform(20, 70), 1), light_lux: round(random.uniform(0, 20000), 0), co2_ppm: round(random.uniform(300, 1200), 0), } API_URL http://127.0.0.1:8000/api/v1/sensors/data if __name__ __main__: while True: payload read_sensor() resp requests.post(API_URL, jsonpayload, timeout10) print(resp.status_code, payload) time.sleep(5)真实接入时把read_sensor()替换成串口读取、Modbus 轮询或 MQTT 订阅即可。上报格式不用变平台侧不用改。6.2 数据上报接口app/routers/sensors.py定义上报和查询接口from datetime import datetime from fastapi import APIRouter, HTTPException from app.database import get_connection from app.schemas import SensorDataIn router APIRouter(prefix/sensors, tags[sensors]) router.post(/data) def upload_data(item: SensorDataIn): ts item.ts or datetime.now().isoformat(timespecseconds) with get_connection() as conn: cur conn.execute( INSERT INTO sensor_data (device_id, ts, air_temp, air_humidity, soil_humidity, light_lux, co2_ppm) VALUES (?, ?, ?, ?, ?, ?, ?) , (item.device_id, ts, item.air_temp, item.air_humidity, item.soil_humidity, item.light_lux, item.co2_ppm)) conn.commit() return {id: cur.lastrowid, ts: ts} router.get(/devices/{device_id}/latest) def latest(device_id: int): with get_connection() as conn: row conn.execute( SELECT * FROM sensor_data WHERE device_id ? ORDER BY ts DESC LIMIT 1 , (device_id,)).fetchone() if row is None: raise HTTPException(status_code404, detailno data) return dict(row) router.get(/timeseries) def timeseries(device_id: int, hours: int 24): with get_connection() as conn: rows conn.execute( SELECT * FROM sensor_data WHERE device_id ? AND ts datetime(now, ?) ORDER BY ts ASC , (device_id, f-{hours} hours)).fetchall() return [dict(r) for r in rows]三个接口对应三种最常用操作POST /api/v1/sensors/data设备上报新数据。GET /api/v1/sensors/devices/{device_id}/latest查某个设备的最新一条数据。GET /api/v1/sensors/timeseries按时间范围查趋势数据供图表使用。主程序app/main.py负责装配服务from pathlib import Path from fastapi import FastAPI from fastapi.responses import HTMLResponse from app.database import init_db from app.routers import alarms, sensors app FastAPI(title农业物联网监测平台, version0.1.0) app.on_event(startup) def on_startup(): init_db() app.include_router(sensors.router, prefix/api/v1) app.include_router(alarms.router, prefix/api/v1) app.get(/health) def health(): return {status: ok} app.get(/dashboard, response_classHTMLResponse) def dashboard(): html_path Path(__file__).parent.parent / dashboard / index.html return html_path.read_text(encodingutf-8)启动服务uvicorn app.main:app --host 0.0.0.0 --port 8000看到Uvicorn running on http://0.0.0.0:8000就说明服务起来了。浏览器访问http://127.0.0.1:8000/docs可以直接看到接口文档这是 FastAPI 自带的能力后续调试非常方便。7. Web 可视化看板数据存进去了接口也能查了下一步用一个 Web 页面把环境趋势展示出来。这里使用 ECharts 绘制折线图页面由 FastAPI 直接返回不引入额外前端框架。dashboard/index.html内容如下!DOCTYPE html html langzh-CN head meta charsetUTF-8 / title大棚 1 号环境监测/title script srchttps://cdn.jsdelivr.net/npm/echarts5/dist/echarts.min.js/script /head body h3大棚 1 号 近 24 小时温湿度趋势/h3 div idchart stylewidth: 100%; height: 480px;/div script async function loadData() { const resp await fetch(/api/v1/sensors/timeseries?device_id1hours24); const rows await resp.json(); const chart echarts.init(document.getElementById(chart)); chart.setOption({ xAxis: { type: category, data: rows.map(r r.ts) }, yAxis: { type: value }, tooltip: { trigger: axis }, series: [ { name: 空气温度, type: line, data: rows.map(r r.air_temp) }, { name: 空气湿度, type: line, data: rows.map(r r.air_humidity) } ] }); } loadData(); setInterval(loadData, 30000); /script /body /html访问http://127.0.0.1:8000/dashboard如果能看到曲线的点随着模拟器上报而增加说明“采集 → 存储 → 查询 → 展示”整条链路已经通了。这里有一个实用的步骤建议先用curl确认接口数据正常再打开看板。这样如果页面空白你马上知道问题在接口层还是前端层。8. 阈值告警与通知环境监测平台不能只做展示还得在指标越界时主动通知。这里实现一个简单的告警检查脚本每隔一段时间扫描最近两分钟的传感器数据发现异常就写入alarm_records表并通过 Webhook 通知。collector/alarm_checker.pyimport sqlite3 import time from pathlib import Path import requests DB_PATH Path(__file__).parent.parent / data / agri_iot.db WEBHOOK_URL https://example.com/hooks/agri-alert ALARM_RULES { air_temp_high: {field: air_temp, operator: , value: 35}, air_temp_low: {field: air_temp, operator: , value: 5}, soil_humidity_low: {field: soil_humidity, operator: , value: 20}, } def check_alarms(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row cursor conn.cursor() rows cursor.execute( SELECT * FROM sensor_data WHERE ts datetime(now, -2 minutes) ).fetchall() for row in rows: for alarm_type, rule in ALARM_RULES.items(): field rule[field] value row[field] if value is None: continue if rule[operator] and value rule[value]: message f设备 {row[device_id]} {field} 越界: {value} cursor.execute( INSERT INTO alarm_records (device_id, type, message) VALUES (?, ?, ?), (row[device_id], alarm_type, message), ) conn.commit() try: requests.post(WEBHOOK_URL, json{text: message}, timeout10) except Exception as exc: print(webhook notify failed:, exc) conn.close() print(alarm check done) if __name__ __main__: while True: check_alarms() time.sleep(60)告警规则用字典集中管理新增规则只需要在ALARM_RULES里加一条不需要改主体逻辑。Webhook 地址可以指向企业内部通知机器人也可以先用一个能收到 POST 请求的测试地址验证。查询告警记录接口放在app/routers/alarms.pyfrom fastapi import APIRouter from app.database import get_connection router APIRouter(prefix/alarms, tags[alarms]) router.get() def list_alarms(limit: int 50): with get_connection() as conn: rows conn.execute( SELECT * FROM alarm_records ORDER BY created_at DESC LIMIT ? , (limit,)).fetchall() return [dict(r) for r in rows]浏览器访问http://127.0.0.1:8000/api/v1/alarms就能看到告警历史。这个接口也可以作为管理后台的数据源后续扩展成告警确认、告警归档都方便。9. 接口 API 调用示例平台服务本身是标准的 REST API所有功能都可以脱离浏览器直接调用。这对设备接入和自动化运维很关键。9.1 上报数据curl -X POST http://127.0.0.1:8000/api/v1/sensors/data \ -H Content-Type: application/json \ -d {device_id:1,air_temp:28.5,air_humidity:60,soil_humidity:45,light_lux:12000,co2_ppm:500}预期返回{ id: 1024, ts: 2026-01-01T10:30:00 }9.2 查询最新数据curl http://127.0.0.1:8000/api/v1/sensors/devices/1/latest9.3 查询时序数据curl http://127.0.0.1:8000/api/v1/sensors/timeseries?device_id1hours249.4 Python 调用方式import requests BASE_URL http://127.0.0.1:8000/api/v1 payload { device_id: 1, air_temp: 26.5, air_humidity: 65, soil_humidity: 40, light_lux: 8000, co2_ppm: 420, } resp requests.post(f{BASE_URL}/sensors/data, jsonpayload, timeout10) print(upload:, resp.status_code, resp.json()) latest requests.get(f{BASE_URL}/sensors/devices/1/latest, timeout10) print(latest:, latest.json())接口能跑通后面就可以把它接到自己的采集脚本、管理后台或第三方系统里。10. 批量任务历史数据回填有了实时数据还需要支持历史数据回填。比如传感器断网一天补传数据时不能一条一条 POST必须走批量导入。scripts/batch_import.py用于把 CSV 文件回填到数据库import csv import sqlite3 from pathlib import Path DB_PATH Path(__file__).parent.parent / data / agri_iot.db CSV_PATH Path(__file__).parent.parent / data / history.csv def import_csv(): conn sqlite3.connect(DB_PATH) with open(CSV_PATH, newline, encodingutf-8) as f: reader csv.DictReader(f) batch [] for row in reader: batch.append(( row[device_id], row[ts], row[air_temp], row[air_humidity], row[soil_humidity], row[light_lux], row[co2_ppm], )) conn.executemany( INSERT INTO sensor_data (device_id, ts, air_temp, air_humidity, soil_humidity, light_lux, co2_ppm) VALUES (?, ?, ?, ?, ?, ?, ?) , batch) conn.commit() conn.close() print(fbatch import done: {len(batch)} rows) if __name__ __main__: import_csv()这里用executemany批量插入比逐条execute快得多。数据量大时还可以在 insert 前先做日期去重避免重复回填。CSV 文件格式参考device_id,ts,air_temp,air_humidity,soil_humidity,light_lux,co2_ppm 1,2026-01-01 00:00:00,18.2,62,35,0,450 1,2026-01-01 00:05:00,18.5,61,35,0,448 1,2026-01-01 00:10:00,18.7,60,34,0,452定时采集这类重复任务可以用cron或APScheduler调度。最简单的做法是让模拟器一直跑采集频率由传感器侧控制平台侧只负责接收和存储不做主动轮询这样接口压力更小。11. 资源占用与性能观察这个案例的资源占用分两部分Claude Code 本身的资源占用以及跑起来的平台服务资源占用。Claude Code 的工作机制是本地终端进程加云端 API 调用。本地只承担文本编辑、命令执行、上下文管理内存占用通常在几百 MB 量级不依赖 GPU。真正需要关注的是 API 调用时的网络时延和 token 消耗。项目上下文越大、要求生成的文件越多单次任务消耗的 token 就越多。建议在交互会话里保持项目文件精简不要一次性丢入无关的大文件。平台服务方面FastAPI 加 SQLite 是轻度资源方案。可以用以下命令观察top或者只看 Python 进程ps aux | grep uvicorn出现高分辨率、大批量回填、多人同时访问时关注三个指标CPU 占用批量导入和大量查询时 CPU 会短暂升高。内存占用SQLite 连接和 FastAPI 进程的内存通常几百 MB。磁盘增长data/agri_iot.db文件会随时间变大需要定期归档或清理。如果要降低数据库体积可以按天分区清理旧数据或者迁移到 InfluxDB 这类时序数据库用保留策略自动降采样。注意具体内存和磁盘占用需要以本机实际数据量为准不同采集频率和字段数量差异很大。12. 接口测试闭环功能都写完建议补一层自动化测试把“上报-查询”这条主链路锁住。tests/test_api.py使用 FastAPI 自带的 TestClientfrom fastapi.testclient import TestClient from app.main import app client TestClient(app) def test_health(): resp client.get(/health) assert resp.status_code 200 assert resp.json()[status] ok def test_upload_and_query(): payload { device_id: 1, air_temp: 28.5, air_humidity: 60, soil_humidity: 45, light_lux: 12000, co2_ppm: 500, } resp client.post(/api/v1/sensors/data, jsonpayload) assert resp.status_code 200 latest client.get(/api/v1/sensors/devices/1/latest).json() assert latest[air_temp] 28.5运行测试pytest tests/test_api.py -v看到PASSED就说明主链路闭环。以后每次改动接口跑一遍测试就能快速发现回归。这里要提醒一点测试环境最好使用独立的测试数据库不要让测试数据污染开发数据。可以给database.py增加环境变量控制数据库路径比如AGRI_IOT_DB。13. 常见问题与排查方法实际部署和开发中最常遇到下面这些问题按表格排查即可。问题现象可能原因排查方式解决方案claude命令提示未授权账号没有对应访问权限或企业策略限制了 Claude Code检查登录状态和账号权限确认账号可用联系管理员开启权限claude提示模型名不被识别配置的模型标识与当前版本支持范围不一致在交互会话中使用/model查看可用模型修改配置使用当前版本支持的模型名npm install报错Node.js 版本过低或网络问题执行node --version检查版本升级 Node.js 到 18 以上后重试pip install失败Python 版本或依赖冲突查看错误日志末尾的包名升级 Python 或单独安装冲突包uvicorn启动后页面打不开端口被占用或服务未启动检查日志执行lsof -i:8000更换端口例如--port 8001POST /api/v1/sensors/data返回 500数据库路径不存在或表未创建查看 FastAPI 日志堆栈确认init_db()在启动时执行看板页面空白接口返回异常或前端 JS 错误打开浏览器开发者工具看 Network/Console先用 curl 调接口确认数据再查前端请求数据不断增多占用磁盘采集频率过高或未归档查询数据库大小增加归档策略定期清理旧数据Webhook 收不到告警通知地址不可达或网络不通手动 curl 测试 Webhook 地址检查地址和网络策略增加失败重试批量导入很慢逐条插入而非批量插入检查脚本是否用了executemany改用批量 insert数据量大时分批提交排查顺序建议固定先看日志再测接口最后查数据库。不要一上来就去改代码。14. 让 Claude Code 继续扩展项目主链路搭完Claude Code 还能继续派上用场。这是本案例最具性价比的部分后续每个功能点都可以作为提示词交给它。交互模式示例请为 sensor_data 表生成一个 CSV 批量导入脚本支持日期去重 并要求在导入前自动初始化数据库表结构。非交互模式适合跑单次任务claude -p 为农业物联网监测平台的 sensor_data 表生成一个 CSV 批量导入脚本使用 Python 标准库 csv输出到 ./scripts/batch_import.py还可以做代码审查检查我这个 FastAPI 项目的安全风险重点看 SQL 注入、缺少鉴权、 路径遍历、敏感信息硬编码。列出问题清单并给出修改建议。再比如增加设备管理、用户登录、MQTT 接入、导出 Excel 报表、对接大屏展示都是自然语言描述需求Claude Code 生成代码你再跑测试验证。整个平台就可以从原型逐步演进成可用系统。15. 最佳实践与合规提醒最后把工程化经验总结一下这些建议适用于大多数 Claude Code 辅助开发的项目。第一先小后大。第一次运行时用模拟数据、低频率、单设备验证链路确认问题出在哪一层再逐步增加设备数和数据量。不要一上来就接几百个真实传感器否则问题排查会非常困难。第二目录和配置分开管理。模型文件、输入素材、输出结果、日志、数据库文件分别放目录敏感配置如 API Key、数据库密码不要写进代码仓库。案例里的数据库默认在data/目录下可以在.gitignore里忽略它。第三批量任务必须有日志和失败重试。采集脚本、告警检查、批量导入都要打印关键步骤Webhook 发送失败时要有重试或落盘机制避免告警丢失。第四接口服务要限制访问范围。开发阶段监听127.0.0.1即可部署到服务器时如果需要对外提供接口必须在前面加鉴权、限流和 HTTPS否则传感器数据会被任意读取。第五涉及人脸、声音、位置、农事经营数据时必须确认授权。本案例没有涉及人脸和声音但农业数据同样有隐私属性采集前要取得相关方同意用于第三方模型或服务前要做脱敏。第六发布或商用前要做效果复核。AI 生成的代码不能直接信任至少要有接口测试、异常输入测试和数据准确性核验。传感器数据的准确性直接影响农业生产判断需要和实际仪器比对后再接入。16. 总结与下一步这个案例最值得尝试的点是用一个下午把“采集-存储-查询-展示-告警”完整跑通而且全程用 Claude Code 辅助编码体验一次 AI 编程代理在真实项目里的工作方式。最先应该验证的功能是主链路把模拟器跑起来然后在http://127.0.0.1:8000/dashboard观察曲线是否更新再用pytest tests/test_api.py确认接口没有回归。这条链路通项目就立住了。最容易踩的坑有两个一是 Claude Code 授权和模型配置问题装完先跑claude --version和/model验证二是端口冲突和数据库初始化失败启动前先确认8000端口空闲启动后确认data/agri_iot.db文件已生成。后续扩展方向很清晰把模拟器换成真实传感器和 MQTT 网关把 SQLite 换成 InfluxDB 并增加保留策略用 Docker 把平台服务和采集器容器化在看板里增加历史对比、设备管理和大屏模式。每一步都可以继续用 Claude Code 辅助开发。这套案例的价值不在于代码本身有多复杂而在于它展示了一个可复制的开发模式用自然语言描述需求让 AI 生成骨架和初版代码再由人补齐测试、排查问题和生产化细节。农业物联网的硬件链路千差万别但软件这条链路用 Claude Code 起步会很快。 
长河君 · 2026/9/2 8:43:42 · 阅读 1.2k 用Claude Code搭建农业物联网监测平台：从传感器到可视化看板
这次我们来看一个非常实际的 AI Agent 应用场景用 Claude Code 从零搭建一个农业物联网监测平台。Claude Code 是 Anthropic 推出的终端 AI 编程代理安装后可以直接在当前项目目录里读写代码、执行命令、跑测试、修 bug。它本身不在本地运行大模型而是通过网络调用云端模型能力所以本机不需要独立显卡也不用下载几十 GB 的模型权重。你只需要一个能跑 Node.js 的终端环境加上有可用权限的账号授权就能开始干活。这篇文章是“Claude Code 100 个案例”系列中的农业物联网案例。目标不是堆概念而是把一条完整链路跑通传感器数据采集 → 数据存储 → REST API → 可视化看板 → 阈值告警 → 历史数据批量回填。整套平台用 FastAPI SQLite 起步单机可跑后续可以平滑换到 PostgreSQL / InfluxDB 这类更强的时间序列存储。读完你会得到几样东西Claude Code 的安装方式和交互用法用提示词让 Claude Code 生成项目骨架的方法一套可以直接复制运行的农业物联网监测平台代码以及接口测试、资源观察和常见问题排查思路。如果你正在做农业信息化、温室大棚监测、果园环境记录或者单纯想看看 Claude Code 这类 AI 编程工具到底能把一个完整项目推进到哪一步这篇文章可以直接收藏。1. 核心能力速览先给结论方便你快速判断这个案例是否值得跟着做。能力项说明项目类型AI 辅助编程 物联网应用开发案例主角工具Claude CodeAnthropic 终端 AI 编程代理平台架构传感器采集 → SQLite 存储 → FastAPI 接口 → Web 看板 → 告警通知主要功能项目脚手架生成、传感器数据上报、时序查询、可视化看板、阈值告警、CSV 批量导入硬件要求普通办公电脑即可建议 8GB 内存以上不需要独立 GPU运行环境Windows / macOS / Linux需要 Node.js 和 Python 3.10Claude Code 启动方式终端交互模式claude非交互模式claude -p平台服务启动方式uvicorn app.main:app启动 Web 服务是否支持 API支持平台提供 RESTful 接口是否支持批量任务支持含定时采集、CSV 历史数据回填数据存储起步用 SQLite生产环境可迁移到 PostgreSQL / InfluxDB适合读者物联网开发者、农业信息化从业者、AI 辅助编程入门者注意一点本文给出的代码是农业物联网平台的常见实现结构属于案例示意。实际让 Claude Code 生成时具体文件名、接口路径、字段命名都会有差异以你本机生成并通过测试的结果为准。2. 适用场景与使用边界这个案例适合谁一句话需要快速验证物联网数据链路的人。不管是温室大棚的空气温湿度、土壤湿度、光照强度还是果园的二氧化碳浓度、气象数据底层的“采集-存储-查询-展示-告警”逻辑是一致的。用 Claude Code 搭建原型能把原来两三天的框架搭建工作压缩到一个下午。但也要说清楚边界。这套平台的定位是原型和中小规模监测不是直接拿去种菜就完事的工业系统。真实生产环境里传感器需要校验、去重、断线续传网关要考虑离线缓存数据库要考虑写入吞吐接口要考虑鉴权、限流、HTTPS。这些在案例里只做了最简版本生产化之前还需要大量加固。另一个必须强调的是数据合规和隐私边界。农业物联网数据可能涉及地块位置、农事记录、产量信息属于经营数据。采集和存储前要确认数据来源合法、相关方已授权不要把未经脱敏的数据上传到未授权的第三方服务如果平台要部署到公网必须给接口加访问控制和身份校验避免传感器数据被任意读取。3. 环境准备与前置条件先列环境清单再逐个确认。3.1 本机工具工具版本建议用途Node.js18 以上安装 Claude Codenpm随 Node.js 自带包管理器Python3.10 以上运行 FastAPI 平台服务Git有即可代码版本管理浏览器任意现代浏览器访问可视化看板确认命令node --version npm --version python --version如果输出正常环境就算准备好了。3.2 安装 Claude CodeClaude Code 的安装很简单用 npm 全局安装npm install -g anthropic-ai/claude-code安装完成后检查版本claude --version第一次运行claude会进入账号授权流程需要 Claude Code 的使用权限和有效登录凭证。如果在受管环境中使用注意企业策略可能限制订阅访问遇到授权类提示时优先找管理员确认账号是否有对应权限。3.3 安装 Python 依赖平台服务部分使用 FastAPI进入项目目录后创建虚拟环境并安装依赖mkdir agri-iot cd agri-iot python -m venv .venv # Windows PowerShell 执行 # .venv\Scripts\activate # macOS / Linux 执行 source .venv/bin/activaterequirements.txt内容如下fastapi0.110 uvicorn[standard]0.29 pydantic2.6 requests2.31 apscheduler3.10 pytest8.0安装pip install -r requirements.txtClaude Code 的核心功能是写代码、改代码、执行命令不是本地推理所以这里没有 CUDA、PyTorch、显卡驱动这些前置依赖。资源瓶颈在 Web 服务和数据库上普通电脑完全带得动。4. 用 Claude Code 设计项目架构装好环境后先不急着写代码让 Claude Code 帮我们把架构理清楚。进入项目目录启动交互模式claude在交互会话里输入请作为资深 Python 工程师帮我设计一个农业物联网监测平台的项目结构。 要求使用 FastAPI SQLite包含传感器数据上报、时序查询、告警记录、 Web 可视化看板、CSV 批量导入。目录结构要清晰模块职责要单一。 先给出整体架构说明再列目录树。Claude Code 会先输出一段架构说明然后给出类似下面的目录结构。不同版本生成结果会有差异但整体思路应该一致agri-iot/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口与路由注册 │ ├── database.py # SQLite 连接与建表 │ ├── models.py # 数据库表结构定义 │ ├── schemas.py # Pydantic 请求/响应模型 │ └── routers/ │ ├── __init__.py │ ├── sensors.py # 传感器数据接口 │ └── alarms.py # 告警记录接口 ├── collector/ │ ├── simulator.py # 模拟传感器 │ └── uploader.py # 数据上报客户端 ├── dashboard/ │ └── index.html # 可视化看板 ├── scripts/ │ ├── init_db.py # 初始化数据库 │ └── batch_import.py # CSV 批量导入 ├── tests/ │ └── test_api.py # 接口测试 ├── requirements.txt └── README.md这个设计的核心思路是分层collector负责采集设备侧数据app负责平台服务dashboard负责展示scripts负责运维类任务。每层之间只通过 HTTP 接口或数据库交互互不耦合。如果你对目录不满意可以直接在会话里追加要求例如把 dashboard 改为用 FastAPI 直接返回单页 HTML不要单独起一个前端服务。Claude Code 会基于对话上下文调整文件结构这是它比普通代码补全工具强的地方。5. 环境初始化与数据库建表架构确定后让 Claude Code 生成数据库模块。这里提供一个可直接使用的app/database.py作用是连接 SQLite 并建表import sqlite3 from pathlib import Path DB_PATH Path(__file__).parent.parent / data / agri_iot.db def get_connection(): DB_PATH.parent.mkdir(parentsTrue, exist_okTrue) conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn def init_db(): with get_connection() as conn: conn.executescript( CREATE TABLE IF NOT EXISTS devices ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, location TEXT, created_at TEXT DEFAULT (datetime(now)) ); CREATE TABLE IF NOT EXISTS sensor_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id INTEGER NOT NULL, ts TEXT NOT NULL, air_temp REAL, air_humidity REAL, soil_humidity REAL, light_lux REAL, co2_ppm REAL, FOREIGN KEY (device_id) REFERENCES devices(id) ); CREATE TABLE IF NOT EXISTS alarm_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id INTEGER, type TEXT, message TEXT, created_at TEXT DEFAULT (datetime(now)) ); )三张表的职责很清楚devices维护设备档案比如“三号大棚”“东区果园”。sensor_data是核心时序数据表每条记录对应一次传感器上报时间字段ts用来做趋势查询。alarm_records保存告警历史方便回看某个时段发生了多少次温度越界。这里用 SQLite 是为了让案例开箱即跑。数据结构稳定后如果要上生产可以把这个建表语句平滑迁移到 PostgreSQL 或 InfluxDB。app/schemas.py定义接口的数据结构from pydantic import BaseModel from typing import Optional class SensorDataIn(BaseModel): device_id: int ts: Optional[str] None air_temp: Optional[float] None air_humidity: Optional[float] None soil_humidity: Optional[float] None light_lux: Optional[float] None co2_ppm: Optional[float] None字段全部设为可选是为了兼容不同传感器上报的字段差异。有的设备只有温湿度有的设备有光照和二氧化碳接收端不需要为每种设备单独写接口。6. 传感器数据采集与设备接入数据接入是物联网平台的第一步。真实场景中传感器通过 Modbus、RS485、LoRa、4G 网关等协议接入案例里我们先用一个模拟器代替物理设备验证整条链条是否通畅。6.1 模拟传感器collector/simulator.py负责每隔几秒生成一条环境数据并上报到平台接口import random import time import requests def read_sensor(): return { device_id: 1, air_temp: round(random.uniform(15, 35), 1), air_humidity: round(random.uniform(40, 80), 1), soil_humidity: round(random.uniform(20, 70), 1), light_lux: round(random.uniform(0, 20000), 0), co2_ppm: round(random.uniform(300, 1200), 0), } API_URL http://127.0.0.1:8000/api/v1/sensors/data if __name__ __main__: while True: payload read_sensor() resp requests.post(API_URL, jsonpayload, timeout10) print(resp.status_code, payload) time.sleep(5)真实接入时把read_sensor()替换成串口读取、Modbus 轮询或 MQTT 订阅即可。上报格式不用变平台侧不用改。6.2 数据上报接口app/routers/sensors.py定义上报和查询接口from datetime import datetime from fastapi import APIRouter, HTTPException from app.database import get_connection from app.schemas import SensorDataIn router APIRouter(prefix/sensors, tags[sensors]) router.post(/data) def upload_data(item: SensorDataIn): ts item.ts or datetime.now().isoformat(timespecseconds) with get_connection() as conn: cur conn.execute( INSERT INTO sensor_data (device_id, ts, air_temp, air_humidity, soil_humidity, light_lux, co2_ppm) VALUES (?, ?, ?, ?, ?, ?, ?) , (item.device_id, ts, item.air_temp, item.air_humidity, item.soil_humidity, item.light_lux, item.co2_ppm)) conn.commit() return {id: cur.lastrowid, ts: ts} router.get(/devices/{device_id}/latest) def latest(device_id: int): with get_connection() as conn: row conn.execute( SELECT * FROM sensor_data WHERE device_id ? ORDER BY ts DESC LIMIT 1 , (device_id,)).fetchone() if row is None: raise HTTPException(status_code404, detailno data) return dict(row) router.get(/timeseries) def timeseries(device_id: int, hours: int 24): with get_connection() as conn: rows conn.execute( SELECT * FROM sensor_data WHERE device_id ? AND ts datetime(now, ?) ORDER BY ts ASC , (device_id, f-{hours} hours)).fetchall() return [dict(r) for r in rows]三个接口对应三种最常用操作POST /api/v1/sensors/data设备上报新数据。GET /api/v1/sensors/devices/{device_id}/latest查某个设备的最新一条数据。GET /api/v1/sensors/timeseries按时间范围查趋势数据供图表使用。主程序app/main.py负责装配服务from pathlib import Path from fastapi import FastAPI from fastapi.responses import HTMLResponse from app.database import init_db from app.routers import alarms, sensors app FastAPI(title农业物联网监测平台, version0.1.0) app.on_event(startup) def on_startup(): init_db() app.include_router(sensors.router, prefix/api/v1) app.include_router(alarms.router, prefix/api/v1) app.get(/health) def health(): return {status: ok} app.get(/dashboard, response_classHTMLResponse) def dashboard(): html_path Path(__file__).parent.parent / dashboard / index.html return html_path.read_text(encodingutf-8)启动服务uvicorn app.main:app --host 0.0.0.0 --port 8000看到Uvicorn running on http://0.0.0.0:8000就说明服务起来了。浏览器访问http://127.0.0.1:8000/docs可以直接看到接口文档这是 FastAPI 自带的能力后续调试非常方便。7. Web 可视化看板数据存进去了接口也能查了下一步用一个 Web 页面把环境趋势展示出来。这里使用 ECharts 绘制折线图页面由 FastAPI 直接返回不引入额外前端框架。dashboard/index.html内容如下!DOCTYPE html html langzh-CN head meta charsetUTF-8 / title大棚 1 号环境监测/title script srchttps://cdn.jsdelivr.net/npm/echarts5/dist/echarts.min.js/script /head body h3大棚 1 号 近 24 小时温湿度趋势/h3 div idchart stylewidth: 100%; height: 480px;/div script async function loadData() { const resp await fetch(/api/v1/sensors/timeseries?device_id1hours24); const rows await resp.json(); const chart echarts.init(document.getElementById(chart)); chart.setOption({ xAxis: { type: category, data: rows.map(r r.ts) }, yAxis: { type: value }, tooltip: { trigger: axis }, series: [ { name: 空气温度, type: line, data: rows.map(r r.air_temp) }, { name: 空气湿度, type: line, data: rows.map(r r.air_humidity) } ] }); } loadData(); setInterval(loadData, 30000); /script /body /html访问http://127.0.0.1:8000/dashboard如果能看到曲线的点随着模拟器上报而增加说明“采集 → 存储 → 查询 → 展示”整条链路已经通了。这里有一个实用的步骤建议先用curl确认接口数据正常再打开看板。这样如果页面空白你马上知道问题在接口层还是前端层。8. 阈值告警与通知环境监测平台不能只做展示还得在指标越界时主动通知。这里实现一个简单的告警检查脚本每隔一段时间扫描最近两分钟的传感器数据发现异常就写入alarm_records表并通过 Webhook 通知。collector/alarm_checker.pyimport sqlite3 import time from pathlib import Path import requests DB_PATH Path(__file__).parent.parent / data / agri_iot.db WEBHOOK_URL https://example.com/hooks/agri-alert ALARM_RULES { air_temp_high: {field: air_temp, operator: , value: 35}, air_temp_low: {field: air_temp, operator: , value: 5}, soil_humidity_low: {field: soil_humidity, operator: , value: 20}, } def check_alarms(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row cursor conn.cursor() rows cursor.execute( SELECT * FROM sensor_data WHERE ts datetime(now, -2 minutes) ).fetchall() for row in rows: for alarm_type, rule in ALARM_RULES.items(): field rule[field] value row[field] if value is None: continue if rule[operator] and value rule[value]: message f设备 {row[device_id]} {field} 越界: {value} cursor.execute( INSERT INTO alarm_records (device_id, type, message) VALUES (?, ?, ?), (row[device_id], alarm_type, message), ) conn.commit() try: requests.post(WEBHOOK_URL, json{text: message}, timeout10) except Exception as exc: print(webhook notify failed:, exc) conn.close() print(alarm check done) if __name__ __main__: while True: check_alarms() time.sleep(60)告警规则用字典集中管理新增规则只需要在ALARM_RULES里加一条不需要改主体逻辑。Webhook 地址可以指向企业内部通知机器人也可以先用一个能收到 POST 请求的测试地址验证。查询告警记录接口放在app/routers/alarms.pyfrom fastapi import APIRouter from app.database import get_connection router APIRouter(prefix/alarms, tags[alarms]) router.get() def list_alarms(limit: int 50): with get_connection() as conn: rows conn.execute( SELECT * FROM alarm_records ORDER BY created_at DESC LIMIT ? , (limit,)).fetchall() return [dict(r) for r in rows]浏览器访问http://127.0.0.1:8000/api/v1/alarms就能看到告警历史。这个接口也可以作为管理后台的数据源后续扩展成告警确认、告警归档都方便。9. 接口 API 调用示例平台服务本身是标准的 REST API所有功能都可以脱离浏览器直接调用。这对设备接入和自动化运维很关键。9.1 上报数据curl -X POST http://127.0.0.1:8000/api/v1/sensors/data \ -H Content-Type: application/json \ -d {device_id:1,air_temp:28.5,air_humidity:60,soil_humidity:45,light_lux:12000,co2_ppm:500}预期返回{ id: 1024, ts: 2026-01-01T10:30:00 }9.2 查询最新数据curl http://127.0.0.1:8000/api/v1/sensors/devices/1/latest9.3 查询时序数据curl http://127.0.0.1:8000/api/v1/sensors/timeseries?device_id1hours249.4 Python 调用方式import requests BASE_URL http://127.0.0.1:8000/api/v1 payload { device_id: 1, air_temp: 26.5, air_humidity: 65, soil_humidity: 40, light_lux: 8000, co2_ppm: 420, } resp requests.post(f{BASE_URL}/sensors/data, jsonpayload, timeout10) print(upload:, resp.status_code, resp.json()) latest requests.get(f{BASE_URL}/sensors/devices/1/latest, timeout10) print(latest:, latest.json())接口能跑通后面就可以把它接到自己的采集脚本、管理后台或第三方系统里。10. 批量任务历史数据回填有了实时数据还需要支持历史数据回填。比如传感器断网一天补传数据时不能一条一条 POST必须走批量导入。scripts/batch_import.py用于把 CSV 文件回填到数据库import csv import sqlite3 from pathlib import Path DB_PATH Path(__file__).parent.parent / data / agri_iot.db CSV_PATH Path(__file__).parent.parent / data / history.csv def import_csv(): conn sqlite3.connect(DB_PATH) with open(CSV_PATH, newline, encodingutf-8) as f: reader csv.DictReader(f) batch [] for row in reader: batch.append(( row[device_id], row[ts], row[air_temp], row[air_humidity], row[soil_humidity], row[light_lux], row[co2_ppm], )) conn.executemany( INSERT INTO sensor_data (device_id, ts, air_temp, air_humidity, soil_humidity, light_lux, co2_ppm) VALUES (?, ?, ?, ?, ?, ?, ?) , batch) conn.commit() conn.close() print(fbatch import done: {len(batch)} rows) if __name__ __main__: import_csv()这里用executemany批量插入比逐条execute快得多。数据量大时还可以在 insert 前先做日期去重避免重复回填。CSV 文件格式参考device_id,ts,air_temp,air_humidity,soil_humidity,light_lux,co2_ppm 1,2026-01-01 00:00:00,18.2,62,35,0,450 1,2026-01-01 00:05:00,18.5,61,35,0,448 1,2026-01-01 00:10:00,18.7,60,34,0,452定时采集这类重复任务可以用cron或APScheduler调度。最简单的做法是让模拟器一直跑采集频率由传感器侧控制平台侧只负责接收和存储不做主动轮询这样接口压力更小。11. 资源占用与性能观察这个案例的资源占用分两部分Claude Code 本身的资源占用以及跑起来的平台服务资源占用。Claude Code 的工作机制是本地终端进程加云端 API 调用。本地只承担文本编辑、命令执行、上下文管理内存占用通常在几百 MB 量级不依赖 GPU。真正需要关注的是 API 调用时的网络时延和 token 消耗。项目上下文越大、要求生成的文件越多单次任务消耗的 token 就越多。建议在交互会话里保持项目文件精简不要一次性丢入无关的大文件。平台服务方面FastAPI 加 SQLite 是轻度资源方案。可以用以下命令观察top或者只看 Python 进程ps aux | grep uvicorn出现高分辨率、大批量回填、多人同时访问时关注三个指标CPU 占用批量导入和大量查询时 CPU 会短暂升高。内存占用SQLite 连接和 FastAPI 进程的内存通常几百 MB。磁盘增长data/agri_iot.db文件会随时间变大需要定期归档或清理。如果要降低数据库体积可以按天分区清理旧数据或者迁移到 InfluxDB 这类时序数据库用保留策略自动降采样。注意具体内存和磁盘占用需要以本机实际数据量为准不同采集频率和字段数量差异很大。12. 接口测试闭环功能都写完建议补一层自动化测试把“上报-查询”这条主链路锁住。tests/test_api.py使用 FastAPI 自带的 TestClientfrom fastapi.testclient import TestClient from app.main import app client TestClient(app) def test_health(): resp client.get(/health) assert resp.status_code 200 assert resp.json()[status] ok def test_upload_and_query(): payload { device_id: 1, air_temp: 28.5, air_humidity: 60, soil_humidity: 45, light_lux: 12000, co2_ppm: 500, } resp client.post(/api/v1/sensors/data, jsonpayload) assert resp.status_code 200 latest client.get(/api/v1/sensors/devices/1/latest).json() assert latest[air_temp] 28.5运行测试pytest tests/test_api.py -v看到PASSED就说明主链路闭环。以后每次改动接口跑一遍测试就能快速发现回归。这里要提醒一点测试环境最好使用独立的测试数据库不要让测试数据污染开发数据。可以给database.py增加环境变量控制数据库路径比如AGRI_IOT_DB。13. 常见问题与排查方法实际部署和开发中最常遇到下面这些问题按表格排查即可。问题现象可能原因排查方式解决方案claude命令提示未授权账号没有对应访问权限或企业策略限制了 Claude Code检查登录状态和账号权限确认账号可用联系管理员开启权限claude提示模型名不被识别配置的模型标识与当前版本支持范围不一致在交互会话中使用/model查看可用模型修改配置使用当前版本支持的模型名npm install报错Node.js 版本过低或网络问题执行node --version检查版本升级 Node.js 到 18 以上后重试pip install失败Python 版本或依赖冲突查看错误日志末尾的包名升级 Python 或单独安装冲突包uvicorn启动后页面打不开端口被占用或服务未启动检查日志执行lsof -i:8000更换端口例如--port 8001POST /api/v1/sensors/data返回 500数据库路径不存在或表未创建查看 FastAPI 日志堆栈确认init_db()在启动时执行看板页面空白接口返回异常或前端 JS 错误打开浏览器开发者工具看 Network/Console先用 curl 调接口确认数据再查前端请求数据不断增多占用磁盘采集频率过高或未归档查询数据库大小增加归档策略定期清理旧数据Webhook 收不到告警通知地址不可达或网络不通手动 curl 测试 Webhook 地址检查地址和网络策略增加失败重试批量导入很慢逐条插入而非批量插入检查脚本是否用了executemany改用批量 insert数据量大时分批提交排查顺序建议固定先看日志再测接口最后查数据库。不要一上来就去改代码。14. 让 Claude Code 继续扩展项目主链路搭完Claude Code 还能继续派上用场。这是本案例最具性价比的部分后续每个功能点都可以作为提示词交给它。交互模式示例请为 sensor_data 表生成一个 CSV 批量导入脚本支持日期去重 并要求在导入前自动初始化数据库表结构。非交互模式适合跑单次任务claude -p 为农业物联网监测平台的 sensor_data 表生成一个 CSV 批量导入脚本使用 Python 标准库 csv输出到 ./scripts/batch_import.py还可以做代码审查检查我这个 FastAPI 项目的安全风险重点看 SQL 注入、缺少鉴权、 路径遍历、敏感信息硬编码。列出问题清单并给出修改建议。再比如增加设备管理、用户登录、MQTT 接入、导出 Excel 报表、对接大屏展示都是自然语言描述需求Claude Code 生成代码你再跑测试验证。整个平台就可以从原型逐步演进成可用系统。15. 最佳实践与合规提醒最后把工程化经验总结一下这些建议适用于大多数 Claude Code 辅助开发的项目。第一先小后大。第一次运行时用模拟数据、低频率、单设备验证链路确认问题出在哪一层再逐步增加设备数和数据量。不要一上来就接几百个真实传感器否则问题排查会非常困难。第二目录和配置分开管理。模型文件、输入素材、输出结果、日志、数据库文件分别放目录敏感配置如 API Key、数据库密码不要写进代码仓库。案例里的数据库默认在data/目录下可以在.gitignore里忽略它。第三批量任务必须有日志和失败重试。采集脚本、告警检查、批量导入都要打印关键步骤Webhook 发送失败时要有重试或落盘机制避免告警丢失。第四接口服务要限制访问范围。开发阶段监听127.0.0.1即可部署到服务器时如果需要对外提供接口必须在前面加鉴权、限流和 HTTPS否则传感器数据会被任意读取。第五涉及人脸、声音、位置、农事经营数据时必须确认授权。本案例没有涉及人脸和声音但农业数据同样有隐私属性采集前要取得相关方同意用于第三方模型或服务前要做脱敏。第六发布或商用前要做效果复核。AI 生成的代码不能直接信任至少要有接口测试、异常输入测试和数据准确性核验。传感器数据的准确性直接影响农业生产判断需要和实际仪器比对后再接入。16. 总结与下一步这个案例最值得尝试的点是用一个下午把“采集-存储-查询-展示-告警”完整跑通而且全程用 Claude Code 辅助编码体验一次 AI 编程代理在真实项目里的工作方式。最先应该验证的功能是主链路把模拟器跑起来然后在http://127.0.0.1:8000/dashboard观察曲线是否更新再用pytest tests/test_api.py确认接口没有回归。这条链路通项目就立住了。最容易踩的坑有两个一是 Claude Code 授权和模型配置问题装完先跑claude --version和/model验证二是端口冲突和数据库初始化失败启动前先确认8000端口空闲启动后确认data/agri_iot.db文件已生成。后续扩展方向很清晰把模拟器换成真实传感器和 MQTT 网关把 SQLite 换成 InfluxDB 并增加保留策略用 Docker 把平台服务和采集器容器化在看板里增加历史对比、设备管理和大屏模式。每一步都可以继续用 Claude Code 辅助开发。这套案例的价值不在于代码本身有多复杂而在于它展示了一个可复制的开发模式用自然语言描述需求让 AI 生成骨架和初版代码再由人补齐测试、排查问题和生产化细节。农业物联网的硬件链路千差万别但软件这条链路用 Claude Code 起步会很快。