简介：本资源为WMS与WCS系统对接的通信代码示例，聚焦仓储管理系统中WMS向WCS下发任务的JSON报文格式与字段定义，面向自动化立体库、智能仓储方向的开发人员与集成工程师。资源以cmd指令区分入库、出库、移库三类业务，并完整给出seq序号、task_id任务唯一编码、起止站台、起止排/列/层坐标及重量、条码等字段的注释说明，便于读者理解任务调度接口的数据结构。压缩包共约2000个文件，以dll动态库、png图片、cshtml页面、js脚本、cs与css样式文件为主，另含xml、config配置及pdb调试文件，整体约367MB，属于典型.NET Web项目结构。目前已有2355人学习下载，可作为WCS任务下发模块的参考实现，帮助读者快速掌握报文组装、字段含义与接口联调思路。
1. 从 WMS 到 WCS 的任务下发：一份能跑通的代码包到底长什么样
很多做仓储系统的兄弟第一次接触 WCS 对接，都会卡在同一个地方：WMS 里出库单、入库单、波次都跑通了，可任务就是下不到设备侧，堆垛机、AGV、输送线全在原地等。这份代码包解决的就是这个断点——它把 WMS 生成出库/入库任务后，如何组装报文、如何调用 WCS 接口、如何处理回执与重试，完整落成了一套可复现的源码。适合两类人：一是刚接手 WMS-WCS 接口层的后端，二是需要给现场设备联调提供稳定任务流的实施。它不讲仓储理论，只讲任务从 WMS 数据库里被捞出来、变成 WCS 能认的指令、再被确认执行这一条链路。下面按我拆包的顺序，把选型、代码、参数和坑一次说清。
2. 任务模型与接口选型：为什么不是直接写 WCS 库表
2.1 WMS 与 WCS 的职责边界
先把边界划清楚，不然后面代码全是糊的。WMS 管的是「要做什么」——出库单、入库单、库存分配、波次、任务优先级；WCS 管的是「怎么动」——把任务拆成设备可执行的动作序列，调度堆垛机、穿梭车、输送线、AGV。两者之间传递的核心对象就是任务（Task），出库任务和入库任务是两条最常走的链路。
常见做法是 WMS 生成任务后，不直接写 WCS 的库表，而是通过接口下发。原因有三个：一是解耦，WCS 换供应商时 WMS 不用重写；二是可追溯，每次下发都有报文和回执；三是可控，WCS 侧能对任务做校验、限流、排队。直接写库表看起来快，但一旦 WCS 表结构变动或者需要加校验，WMS 就被拖死，这是血泪经验。
这份代码包采用的模型是：WMS 侧维护一张任务表，任务状态从「待下发」到「已下发」到「执行中」到「完成/失败」，WCS 侧提供任务接收接口和状态回传接口。出库任务和入库任务共用一套下发框架，差异在任务类型字段和扩展参数上。
2.2 接口协议与报文格式的取舍
接口选型上，常见的有 REST/JSON、WebService、MQ 消息、TCP 自定义报文。这份代码包用的是 REST/JSON 为主、MQ 为辅的结构：任务下发走 HTTP POST，状态回传走 MQ 或者回调接口。为什么这么选？因为现场联调时 HTTP 最容易抓包和模拟，用 Postman 就能打；而状态回传用 MQ 能避免 WMS 被 WCS 的回调打爆。
报文格式上，出库任务和入库任务的字段有共性也有差异。共性字段包括任务号、任务类型、优先级、容器号、起始位置、目标位置、创建时间；出库任务多一个出库单号和波次号，入库任务多一个入库单号和供应商信息。代码包里把这些抽成一个基类，子类只扩展差异字段，避免每个任务类型写一遍。
提示：接口字段命名一定要和 WCS 供应商对齐，尤其是位置编码格式。我见过因为 WMS 用「A-01-02-03」而 WCS 要「A010203」导致任务全部下不去的情况，联调前先对字段字典。
2.3 任务下发的整体流程
流程拆成六步：WMS 业务触发任务生成、任务落库为待下发、下发服务捞取待下发任务、组装报文调用 WCS 接口、根据回执更新任务状态、失败进入重试队列。这六步在代码里对应六个模块，下面会逐个落到代码。
3. 出库任务下发：从任务组装到接口调用的完整代码
3.1 任务实体与状态机设计
先看任务实体。出库任务和入库任务共用一个 Task 基类，状态用枚举管理，避免到处写魔法数字。
from enum import Enum from dataclasses import dataclass, field from datetime import datetime class TaskType(Enum): OUTBOUND = "OUTBOUND" # 出库任务 INBOUND = "INBOUND" # 入库任务 class TaskStatus(Enum): PENDING = "PENDING" # 待下发 SENT = "SENT" # 已下发，等回执 EXECUTING = "EXECUTING" # WCS 已接收，执行中 DONE = "DONE" # 完成 FAILED = "FAILED" # 失败，可重试 @dataclass class Task: task_no: str # WMS 侧任务号，全局唯一 task_type: TaskType priority: int = 5 # 1 最高，9 最低 container_no: str = "" # 容器/托盘号 from_location: str = "" # 起始位置编码 to_location: str = "" # 目标位置编码 status: TaskStatus = TaskStatus.PENDING retry_count: int = 0 created_at: datetime = field(default_factory=datetime.now) ext: dict = field(default_factory=dict) # 扩展字段，放单号、波次等这段代码的关键点是task_no必须由 WMS 生成且全局唯一，WCS 侧用它做幂等。ext字典用来放差异字段，出库放outbound_no和wave_no，入库放inbound_no和supplier。retry_count是重试次数的计数器，后面重试逻辑会读它。
状态机的流转规则要写死：PENDING 只能到 SENT，SENT 到 EXECUTING 或 FAILED，EXECUTING 到 DONE 或 FAILED，FAILED 可以回到 PENDING 重试。不要允许跳状态，否则现场排查时根本不知道任务卡在哪。
3.2 出库任务组装与下发代码
出库任务的核心是把 WMS 的出库单信息转成 WCS 能认的指令。下面这段是组装和下发的主逻辑。
import requests import json WCS_TASK_URL = "http://wcs-host:8080/api/task/receive" TIMEOUT = 5 # 秒，现场网络抖动大，别设太长 def build_outbound_payload(task: Task) -> dict: """把出库任务组装成 WCS 报文""" return { "taskNo": task.task_no, "taskType": "OUT", "priority": task.priority, "containerNo": task.container_no, "fromLocation": task.from_location, "toLocation": task.to_location, "outboundNo": task.ext.get("outbound_no", ""), "waveNo": task.ext.get("wave_no", ""), "createTime": task.created_at.strftime("%Y-%m-%d %H:%M:%S") } def send_task_to_wcs(task: Task) -> bool: """下发单个任务，返回是否成功""" payload = build_outbound_payload(task) try: resp = requests.post( WCS_TASK_URL, data=json.dumps(payload), headers={"Content-Type": "application/json"}, timeout=TIMEOUT ) if resp.status_code == 200: result = resp.json() # WCS 约定 code=0 表示接收成功 if result.get("code") == 0: task.status = TaskStatus.SENT return True else: # 业务失败，记录 WCS 返回的错误信息 task.ext["wcs_error"] = result.get("msg", "unknown") return False return False except requests.exceptions.Timeout: task.ext["wcs_error"] = "timeout" return False except requests.exceptions.RequestException as e: task.ext["wcs_error"] = str(e) return False逻辑说明：build_outbound_payload负责字段映射，注意taskType用的是 WCS 约定的「OUT」而不是枚举值，这种映射一定要单独抽函数，别散在业务代码里。send_task_to_wcs里区分了 HTTP 层失败和业务层失败，超时单独捕获，因为超时和业务拒绝的处理策略不同——超时可能是任务已经进去了但回执没回来，需要靠幂等查询确认，不能直接重发。
参数说明：WCS_TASK_URL是 WCS 提供的任务接收地址，现场部署时换成实际 IP 和端口。TIMEOUT设 5 秒，是因为 WCS 接收任务通常是写库或入队，正常在毫秒级，超过 5 秒基本是网络或 WCS 卡了。priority直接透传，WCS 侧按这个排序。
3.3 批量下发与并发控制
单个下发跑通后，实际场景是批量。出库波次一放就是几十上百个任务，不能串行一个个发，但也不能无脑并发把 WCS 打挂。
from concurrent.futures import ThreadPoolExecutor, as_completed MAX_WORKERS = 8 # 并发数，按 WCS 承受能力调 def batch_send(tasks: list) -> dict: """批量下发，返回成功和失败列表""" success, failed = [], [] with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: future_map = {executor.submit(send_task_to_wcs, t): t for t in tasks} for future in as_completed(future_map): task = future_map[future] if future.result(): success.append(task.task_no) else: failed.append(task.task_no) return {"success": success, "failed": failed}并发数MAX_WORKERS是现场调出来的，默认 8。WCS 如果是单线程接收，并发高了反而排队超时，这时候要降到 2 到 4。批量下发后，成功和失败要分别落库，失败的进重试队列，不要直接丢弃。
4. 入库任务与状态回传：回执处理、幂等与重试机制
4.1 入库任务的差异处理
入库任务和出库任务共用下发框架，差异在报文组装。入库多的是入库单号、供应商、质检状态，起始位置通常是收货口，目标位置是存储位。
def build_inbound_payload(task: Task) -> dict: """入库任务报文，注意 taskType 和扩展字段""" return { "taskNo": task.task_no, "taskType": "IN", "priority": task.priority, "containerNo": task.container_no, "fromLocation": task.from_location, "toLocation": task.to_location, "inboundNo": task.ext.get("inbound_no", ""), "supplier": task.ext.get("supplier", ""), "qcStatus": task.ext.get("qc_status", "PASS"), "createTime": task.created_at.strftime("%Y-%m-%d %H:%M:%S") }qcStatus是质检状态，入库时如果质检未完成，WCS 可能要把任务送到待检区而不是存储区，这个字段一定要和 WCS 对齐取值。supplier在部分 WCS 里用于分区策略，别漏传。
4.2 状态回传接口与幂等处理
WCS 执行完任务后会回传状态，WMS 侧要提供一个接收接口。这个接口最大的坑是重复回传，所以必须幂等。
def handle_wcs_callback(data: dict) -> dict: """处理 WCS 状态回传，幂等""" task_no = data.get("taskNo") wcs_status = data.get("status") # EXECUTING / DONE / FAILED if not task_no: return {"code": 1, "msg": "taskNo missing"} task = load_task(task_no) if task is None: return {"code": 1, "msg": "task not found"} # 幂等：已完成的任务不再处理 if task.status == TaskStatus.DONE: return {"code": 0, "msg": "already done"} if wcs_status == "EXECUTING": task.status = TaskStatus.EXECUTING elif wcs_status == "DONE": task.status = TaskStatus.DONE elif wcs_status == "FAILED": task.status = TaskStatus.FAILED task.ext["wcs_error"] = data.get("errorMsg", "") save_task(task) return {"code": 0, "msg": "ok"}幂等判断放在最前面，DONE状态直接返回成功，避免重复更新。load_task和save_task是持久层方法，实际项目里换成 ORM 或 DAO。回传接口要记录原始报文，现场扯皮时这是唯一证据。
4.3 重试机制与死信处理
失败任务不能无限重试，要有上限和退避。
MAX_RETRY = 3 RETRY_INTERVAL = [10, 30, 60] # 秒，退避间隔 def retry_failed_tasks(): """捞取失败任务重试""" tasks = query_tasks(status=TaskStatus.FAILED) for task in tasks: if task.retry_count >= MAX_RETRY: mark_dead_letter(task) # 进死信，人工处理 continue task.retry_count += 1 task.status = TaskStatus.PENDING save_task(task) # 实际下发由调度器按 RETRY_INTERVAL 触发MAX_RETRY设 3 是经验值，超过 3 次基本是数据问题不是网络问题，再重试也是浪费。退避间隔按retry_count取，第一次 10 秒，第二次 30 秒，第三次 60 秒。死信任务要能人工干预，现场经常需要手动改位置或容器号后重新下发。
注意：重试前一定要确认 WCS 侧没有已经接收该任务。如果第一次下发是超时但 WCS 实际收到了，重试会导致重复任务。常见做法是重试前先调 WCS 的任务查询接口确认。
5. 避坑与排查：任务下发不成功时先看这几处
5.1 任务一直停在 PENDING 不下发
现象：任务落库了，状态是 PENDING，但下发服务不捞。原因通常是调度器没启动，或者捞取条件写错，比如只捞了出库没捞入库，或者时间条件把新任务过滤掉了。解决：先看调度器日志有没有执行，再检查捞取 SQL 的 where 条件，把状态和时间范围打出来。
5.2 WCS 返回成功但任务没执行
现象：下发接口返回 code=0，但设备不动，任务状态停在 SENT。原因多半是 WCS 接收后入队了但调度没触发，或者位置编码 WCS 不认识导致任务被挂起。解决：让 WCS 侧查任务队列，确认任务是否被消费；同时核对位置编码格式，这是最高频的翻车点。
5.3 重复任务导致设备动作两次
现象：同一个容器被送了两次，或者堆垛机执行了重复指令。原因是没有幂等，超时重试或者 WCS 重复回传都会触发。解决：WMS 侧下发前用 task_no 查重，WCS 侧接收时也做幂等；回传接口对 DONE 状态直接返回成功不重复处理。
5.4 状态回传丢失导致任务永远 EXECUTING
现象：任务卡在 EXECUTING，WCS 说已完成，WMS 没收到回传。原因是 MQ 丢消息或者回调接口报错。解决：加对账机制，定时用 WCS 的任务查询接口拉取状态，和 WMS 本地比对，不一致的以 WCS 为准修正。这是黑匣子最容易被忽略的地方。
5.5 批量下发把 WCS 打挂
现象：波次一放，WCS 接口超时，任务大面积失败。原因是并发太高，WCS 接收能力有限。解决：降MAX_WORKERS，加限流，或者改成 MQ 异步下发让 WCS 自己消费。现场联调时先小批量试，别一上来就放全量波次。
6. 进阶：用对账任务兜住状态不一致，附验证脚本
前面讲的都是正常链路，但现场跑久了，状态不一致是必然的。WMS 和 WCS 各有一套状态，网络抖动、服务重启、MQ 丢消息都会让两边对不上。我一般会加一个对账任务，定时跑，把差异捞出来修正。这是这套代码包里最值钱的部分，因为它决定了系统能不能长期稳定跑。
对账的逻辑是：拉取 WMS 侧所有非终态任务（PENDING、SENT、EXECUTING），逐个调 WCS 的任务查询接口，拿到 WCS 侧状态后比对。WCS 说 DONE 而 WMS 还是 EXECUTING 的，修正为 DONE；WCS 说没有这个任务的，说明下发根本没成功，回退到 PENDING 重试；WCS 说 FAILED 的，同步失败原因。
def reconcile_tasks(): """对账：以 WCS 状态为准修正 WMS""" wms_tasks = query_tasks(status_in=[ TaskStatus.PENDING, TaskStatus.SENT, TaskStatus.EXECUTING ]) fixed = 0 for task in wms_tasks: wcs_info = query_wcs_task(task.task_no) # 调 WCS 查询接口 if wcs_info is None: # WCS 没有该任务，回退重试 task.status = TaskStatus.PENDING task.retry_count += 1 save_task(task) fixed += 1 continue wcs_status = wcs_info.get("status") if wcs_status == "DONE" and task.status != TaskStatus.DONE: task.status = TaskStatus.DONE save_task(task) fixed += 1 elif wcs_status == "FAILED" and task.status != TaskStatus.FAILED: task.status = TaskStatus.FAILED task.ext["wcs_error"] = wcs_info.get("errorMsg", "") save_task(task) fixed += 1 return fixedquery_wcs_task是 WCS 提供的任务查询接口，如果 WCS 没提供，这个方案就落不了地，所以接口选型时一定要把查询接口写进合同。对账频率我一般设 5 分钟一次，太频繁浪费资源，太稀疏差异积累多了修正成本高。对账结果要打日志，修正了多少条、哪些任务被回退，这些数据是判断系统健康度的指标。
验证脚本可以单独跑，不依赖调度器：
# 手动触发一次对账，观察输出 python -m wms_wcs.reconcile --once --verbose # 输出示例：reconcile done, fixed=3, pending_retry=1, done_sync=2跑完看fixed数量，如果每次都是 0，说明链路健康；如果持续大于 0，说明回传链路有问题，要去查 MQ 或回调接口。我习惯在每次现场上线后连续观察三天对账数据，稳定为 0 才算验收通过。
从那以后我每次做 WMS-WCS 对接，都会先把对账脚本写好再写下发逻辑，因为下发只是开始，能长期对得上才是终点。希望帮到你。
本文还有配套的精品资源，点击获取
