搜索确认了关键事实，避免凭空捏造。现在写正文和源码，核心抓手是"转转生态模型与闲鱼本质不同"——官方已有 MCP 工具（回收估价/购前找货），但交易类能力仍偏私有 App 接口，这正是适配层设计的真正价值所在。
🔄《转转开放平台API接入实录：与闲鱼接口模型的异同与适配层设计》（附Python源码）
结论先拍：转转开放平台
open.zhuanzhuan.com 确实存在，但生态定位与闲鱼本质不同——官方已上线 MCP Toolkit，当前开放的是"回收 + 购前找货"两条只读业务路径（识别估价、打开回收入口、搜索在售、查询成交行情、商品详情），定位是"把可信二手能力嵌入开发者产品"；而写侧的完整交易能力（订单、发货、退款）仍以 App 私有接口为主。 这意味着转转的接入模型不能照搬闲鱼：闲鱼是"TOP 体系 + 店铺授权 + 全链路交易 API"，转转是"MCP 工具 + 按场景权限 + 只读为主"。 正确做法是适配层（Anti-Corruption Layer）屏蔽差异，让上层业务代码无感切换——下面给出完整的 ZhuanzhuanAdapter + 与闲鱼的差异对照。一、转转 vs 闲鱼：生态模型根本差异
维度 | 闲鱼（阿里生态） | 转转（58+腾讯生态） |
|---|---|---|
底层体系 | 淘宝开放平台（TOP）OAuth 2.0 | 独立开放平台 + MCP Toolkit |
授权 | OAuth + SessionKey（access_token） | MCP 凭证 + 按场景权限分级 |
核心定位 | 全品类 C2C 轻撮合 + 社区 | 垂直品类 + C2B2C 官方验（重履约） |
已开放能力 | 商品/订单/发货/退款全链路 | 回收估价 + 购前找货（当前5个工具） |
交易写能力 | 完整（publish/ship/refund） | 有限，多为 App 私有接口 |
成色模型 | stuff_status 0~100 int | quality 字符串（"95新"）+ qualityDesc |
信任机制 | 社区信用 + 验货宝（可选） | 平台质检 + 质保 + 7天无理由（内建） |
签名 | MD5（TOP 规则） | MCP 调用 / App zzreqsign（MD5系，含Native层） |
关键洞察：转转的商业基因是 "服务+重履约"（二手3C、奢侈品走官方验、出具质检报告、质保），闲鱼是 "社区+轻撮合"。 这个差异直接映射到字段模型——转转的qualityDesc（质检描述）、attributes（规格结构化）、market_price（成交行情）在闲鱼里要么没有、要么是可选增值。
二、转转 MCP 工具全景（官方现行）
官方
open.zhuanzhuan.com 明确两条路径、5 个工具：路径 | 工具 | 作用 |
|---|---|---|
RECYCLE 回收 | recycle_valuation | 识别估价（图/文 → 回收价） |
open_recycle_entry | 打开回收入口（不代用户下单） | |
PRE-SHOPPING 购前 | search | 搜索在售商品 |
market_price | 成交行情（明确型号最新一期价格区间） | |
product_detail | 商品详情（规格+服务权益+质检） |
重要特征：
只读工具，不产生交易操作（官方原文："不产生交易操作"）
最小权限，按场景申请
沙箱环境隔离
参数与响应 Schema 免登录查看
所以生产级 ERP 若需写操作（发布/发货/退款），转转侧目前只能走 App 私有接口或等官方开放——适配层必须为这种"能力不对等"做好降级设计（闲鱼侧走真实交易 API，转转侧只做估价/行情/只读）。
三、字段模型差异（适配层核心翻译）
3.1 成色：闲鱼 int vs 转转 字符串
# 闲鱼: stuff_status (0~100 int)
{"stuff_status": 95} # 95新
# 转转: quality (字符串) + qualityDesc (质检描述)
{"quality": "95新", "qualityDesc": "屏幕细微划痕,功能完好,无修无拆"}适配层必须双向转换（前篇
IdleItemPublishMapper 的思路反转）。3.2 商品：闲鱼扁平 vs 转转结构化
# 闲鱼: title/price/original_price 平铺
{"title": "...", "reserve_price": "99.00", "original_price": "200.00"}
# 转转: 规格走 attributes[], 带品牌/型号/行情
{"brand": "Apple", "model": "iPhone 13", "attributes": [
{"name": "容量", "value": "128GB"}, ...], "marketPrice": "...",
"qualityDesc": "..."}3.3 业务路径差异
闲鱼: 发布商品 → 买家下单 → 卖家发货 → 退款 (全链路闭环) 转转: 识别估价 → 打开回收入口 → (平台介入) → 官方验 → 出售 购前找货: 搜索 → 行情 → 详情 (辅助决策, 非交易)
四、Python：ZhuanzhuanAdapter（适配层完整实现）
# zhuanzhuan_adapter.py
"""
转转开放平台 MCP 适配层 (与闲鱼接口模型对齐)
- MCP 凭证管理 (替代闲鱼 OAuth/SessionKey)
- 5个工具封装: recycle_valuation / open_recycle_entry / search / market_price / product_detail
- 字段翻译: 转转 quality/attributes ↔ 统一 Product 实体
- 能力探测: 转转只读为主, 写操作自动降级/抛出(与闲鱼不对等)
- 复用前几篇: ComplianceGate / ObservabilityMiddleware / TwoLevelCache
"""
import time, hashlib, json, threading
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from enum import Enum
from abc import ABC, abstractmethod
# ==================== 统一领域实体 (与前篇一致) ====================
class OrderStatus(Enum):
CREATED = "CREATED"; PAID = "PAID"; SHIPPED = "SHIPPED"
SIGNED = "SIGNED"; REFUNDING = "REFUNDING"; CLOSED = "CLOSED"
@dataclass
class Money:
amount: float
currency: str = "CNY"
@dataclass
class Product:
sku_id: str
platform: str
title: str
price: Money
original_price: Money = None
stock: int = 0
# 二手专属
condition_int: int = 0 # 统一成色 (0~100, 闲鱼style)
condition_text: str = "" # 原始描述 (转转quality)
quality_desc: str = "" # 质检描述 (转转独有)
attributes: Dict = field(default_factory=dict) # 规格结构化
market_price_range: Optional[Dict] = None # 行情(转转独有)
@dataclass
class Order:
order_id: str
platform: str
status: OrderStatus
total: Money
items_count: int = 0
raw: Dict = field(default_factory=dict)
# ==================== 仓储端口 (与闲鱼共用同一套接口) ====================
class ProductRepository(ABC):
@abstractmethod
def get_product(self, shop_id: str, sku_id: str) -> Optional[Product]: ...
@abstractmethod
def update_stock(self, shop_id: str, sku_id: str, qty: int) -> bool: ...
@abstractmethod
def search(self, shop_id: str, keyword: str, **kw) -> List[Product]: ...
class OrderRepository(ABC):
@abstractmethod
def list_orders(self, shop_id: str, since) -> List[Order]: ...
@abstractmethod
def get_order(self, shop_id: str, order_id: str) -> Optional[Order]: ...
# ==================== 异常 ====================
class ZhuanzhuanError(Exception): pass
class CapabilityNotAvailable(ZhuanzhuanError):
"""转转该能力未开放(只读/App私有), 适配层降级信号"""
pass
# ==================== MCP 凭证 (替代OAuth) ====================
@dataclass
class McpCredential:
"""转转 MCP 访问凭证 (官方: 沙箱隔离 + 权限分级)"""
mcp_token: str
tool_scope: List[str] = field(default_factory=lambda: [
"recycle_valuation", "open_recycle_entry",
"search", "market_price", "product_detail"])
sandbox: bool = True
def has(self, tool: str) -> bool:
return tool in self.tool_scope
# ==================== 字段翻译器 ====================
class ZhuanzhuanFieldTranslator:
"""转转字段 ↔ 统一实体 (核心: 成色/规格/行情)"""
# 转转 quality 字符串 → 0~100 int (对齐闲鱼 stuff_status)
QUALITY_TO_INT = {
"全新": 10, "全新未拆封": 100, "未拆封": 100,
"准新": -1, "近全新": -1,
"99新": 99, "九九新": 99, "9.9新": 99,
"95新": 95, "九五新": 95,
"9成新": 9, "90新": 9,
"85新": 85, "八五新": 85,
"8成新": 8, "80新": 8,
"7成新": 7, "70新": 7,
}
def quality_to_int(self, quality: str) -> int:
if quality is None: return 0
s = str(quality).strip()
if s.lstrip("-").isdigit():
v = int(s)
return v if v == -1 or 0 <= v <= 100 else 0
return self.QUALITY_TO_INT.get(s, 0)
def to_unified(self, raw: Dict, platform: str = "zhuanzhuan") -> Product:
"""转转原始 → 统一 Product"""
attrs = {a["name"]: a["value"] for a in raw.get("attributes", [])}
mp = raw.get("marketPrice")
mpr = None
if mp:
mpr = {"low": self._to_float(mp.get("low")),
"high": self._to_float(mp.get("high"))}
return Product(
sku_id=str(raw.get("itemId") or raw.get("productId") or ""),
platform=platform,
title=raw.get("title", ""),
price=Money(self._to_float(raw.get("price"))),
original_price=Money(self._to_float(raw.get("originalPrice")) or
self._to_float(raw.get("price"))),
stock=0, # 转转只读, 无库存概念(官方验由平台控)
condition_int=self.quality_to_int(raw.get("quality")),
condition_text=raw.get("quality", ""),
quality_desc=raw.get("qualityDesc", ""),
attributes=attrs,
market_price_range=mpr,
)
def from_unified(self, product: Product) -> Dict:
"""统一 Product → 转转发布结构 (若开放写时复用映射器逻辑)"""
return {
"title": product.title,
"price": f"{product.price.amount:.2f}",
"originalPrice": f"{(product.original_price or product.price).amount:.2f}",
"quality": product.condition_text or self._int_to_quality(product.condition_int),
"qualityDesc": product.quality_desc,
"attributes": [{"name": k, "value": v} for k, v in product.attributes.items()],
}
def _int_to_quality(self, v: int) -> str:
inv = {iv: k for k, iv in self.QUALITY_TO_INT.items()}
return inv.get(v, "9成新")
def _to_float(self, x) -> float:
try: return float(x)
except (TypeError, ValueError): return 0.0
# ==================== 转转 MCP 客户端 ====================
class ZhuanzhuanMcpClient:
"""MCP Toolkit 调用封装 (官方: 沙箱隔离, 权限分级)"""
def __init__(self, cred: McpCredential, translator: ZhuanzhuanFieldTranslator,
gateway: str = "https://open.zhuanzhuan.com/mcp"):
self.cred = cred
self.t = translator
self.gateway = gateway
self._cache = {} # 简化: 生产用 TwoLevelCache
def _invoke(self, tool: str, params: Dict) -> Dict:
"""MCP 调用 (演示: 返回模拟数据; 生产走 MCP 协议)"""
if not self.cred.has(tool):
raise CapabilityNotAvailable(f"工具 {tool} 不在权限范围: {self.cred.tool_scope}")
# 模拟响应 (结构对齐官方 Schema)
return self._mock(tool, params)
def _mock(self, tool: str, p: Dict) -> Dict:
if tool == "recycle_valuation":
return {"code": 200, "data": {"estimatedPrice": "1800",
"qualityDesc": "屏幕细微划痕,功能完好", "quality": "90新"}}
if tool == "open_recycle_entry":
return {"code": 200, "data": {"entryUrl": "https://recycle.zhuanzhuan.com/xxx",
"note": "已打开入口, 不代用户下单"}}
if tool == "search":
return {"code": 200, "data": {"items": [
{"productId": "p001", "title": "95新 iPhone13", "price": "2200",
"quality": "95新", "brand": "Apple"}]}}
if tool == "market_price":
return {"code": 200, "data": {"low": "2000", "high": "2600",
"model": p.get("model", "")}}
if tool == "product_detail":
return {"code": 200, "data": {"productId": p.get("productId"),
"title": "95新 iPhone13 128G", "price": "2300",
"quality": "95新", "qualityDesc": "全原装无修",
"attributes": [{"name": "容量", "value": "128GB"}],
"marketPrice": {"low": "2000", "high": "2600"}}}
return {"code": 200, "data": {}}
# ==================== 适配层主体 (实现统一端口) ====================
class ZhuanzhuanAdapter(ProductRepository, OrderRepository):
"""转转适配层: 对上层暴露与闲鱼完全一致的端口"""
PLATFORM = "zhuanzhuan"
def __init__(self, cred: McpCredential, client: ZhuanzhuanMcpClient):
self.cred = cred
self.client = client
# ---- ProductRepository ----
def get_product(self, shop_id: str, sku_id: str) -> Optional[Product]:
raw = self.client._invoke("product_detail", {"productId": sku_id})
if raw.get("code") != 200: return None
return self.client.t.to_unified(raw["data"], self.PLATFORM)
def search(self, shop_id: str, keyword: str, **kw) -> List[Product]:
raw = self.client._invoke("search", {"keyword": keyword, **kw})
return [self.client.t.to_unified(it, self.PLATFORM)
for it in raw.get("data", {}).get("items", [])]
def update_stock(self, shop_id: str, sku_id: str, qty: int) -> bool:
# ★ 转转官方验模式: 库存由平台质检中心控制, ERP 侧只读
raise CapabilityNotAvailable(
"转转官方验(C2B2C)库存由平台管控, 不支持商家直接改库存; "
"若需回写请走官方开放的交易写接口(当前未开放)")
# ---- OrderRepository (转转侧多为App私有, 明确降级) ----
def list_orders(self, shop_id: str, since) -> List[Order]:
raise CapabilityNotAvailable(
"转转订单列表未纳入 MCP 开放工具, 需走 App 私有接口或等待官方开放; "
"建议转转侧仅做只读行情/估价, 交易主链路走闲鱼")
def get_order(self, shop_id: str, order_id: str) -> Optional[Order]:
raise CapabilityNotAvailable("转转订单详情未开放(同上)")
# ---- 转转特色能力 (行情/估价, 闲鱼侧无对应) ----
def estimate_recycle(self, shop_id: str, image_or_text: str) -> Dict:
"""回收估价 (转转独有, 对齐官方 recycle_valuation)"""
raw = self.client._invoke("recycle_valuation", {"input": image_or_text})
return self.client.t.to_unified({
"price": raw["data"]["estimatedPrice"],
"quality": raw["data"].get("quality", ""),
"qualityDesc": raw["data"].get("qualityDesc", ""),
}, self.PLATFORM).__dict__
def market_price(self, shop_id: str, model: str) -> Optional[Dict]:
"""成交行情 (转转独有, 对齐 market_price)"""
raw = self.client._invoke("market_price", {"model": model})
if raw.get("code") != 200: return None
d = raw["data"]
return {"low": self.client.t._to_float(d.get("low")),
"high": self.client.t._to_float(d.get("high")), "model": d.get("model")}
def open_recycle(self, shop_id: str) -> str:
"""打开回收入口 (不代下单, 对齐 open_recycle_entry)"""
raw = self.client._invoke("open_recycle_entry", {})
return raw["data"]["entryUrl"]
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex 注册链接
# ==================== 能力探测器 (能力不对等的核心) ====================
class CapabilityProbe:
"""探测各平台能力, 业务侧按结果分支(不抛异常)"""
FEATURES = ["product_read", "product_write", "stock_write",
"order_read", "order_write", "ship", "refund",
"valuation", "market_price"] # 转转独有: valuation/market_price
def __init__(self, adapters: Dict[str, Any]):
self.adapters = adapters
def probe(self, platform: str) -> Dict[str, bool]:
a = self.adapters.get(platform)
if platform == "zhuanzhuan":
return {
"product_read": True, # MCP 支持
"product_write": False, # 未开放
"stock_write": False, # 平台控库存
"order_read": False, # App私有
"order_write": False,
"ship": False,
"refund": False,
"valuation": True, # ★ 转转独有
"market_price": True, # ★ 转转独有
}
if platform == "idle": # 闲鱼 (前篇能力)
return {
"product_read": True, "product_write": True,
"stock_write": True, "order_read": True,
"order_write": True, "ship": True, "refund": True,
"valuation": False, "market_price": False,
}
return {f: False for f in self.FEATURES}
def choose_platform(self, feature: str) -> List[str]:
"""按能力选平台: 如 'valuation' → ['zhuanzhuan']"""
return [p for p, cap in ((k, self.probe(k)) for k in self.adapters)
if cap.get(feature)]
# ==================== 演示 ====================
if __name__ == "__main__":
cred = McpCredential(mcp_token="test_mcp_token", sandbox=True)
translator = ZhuanzhuanFieldTranslator()
client = ZhuanzhuanMcpClient(cred, translator)
adapter = ZhuanzhuanAdapter(cred, client)
print("=== 转转 MCP: 商品详情 → 统一 Product ===")
p = adapter.get_product("shop_zz", "p001")
print(f" title={p.title} price={p.price.amount} "
f"condition_int={p.condition_int} condition_text={p.condition_text}")
print(f" quality_desc={p.quality_desc} attrs={p.attributes} "
f"market_range={p.market_price_range}")
print("\n=== 转转独有: 回收估价 ===")
est = adapter.estimate_recycle("shop_zz", "iPhone13 128G 蓝色")
print(f" {est}")
print("\n=== 转转独有: 成交行情 ===")
mp = adapter.market_price("shop_zz", "iPhone 13")
print(f" {mp}")
print("\n=== 转转独有: 打开回收入口 ===")
url = adapter.open_recycle("shop_zz")
print(f" {url}")
print("\n=== 能力不对等: 转转不支持写 ===")
try:
adapter.update_stock("shop_zz", "p001", 5)
except CapabilityNotAvailable as e:
print(f" ⚠️ {e}")
try:
adapter.list_orders("shop_zz", None)
except CapabilityNotAvailable as e:
print(f" ⚠️ {e}")
print("\n=== 能力探测: 按能力选平台 ===")
probe = CapabilityProbe({"zhuanzhuan": adapter, "idle": None})
for feat in ["valuation", "market_price", "ship", "product_write"]:
print(f" {feat}: 推荐平台 = {probe.choose_platform(feat)}")
print("\n=== 字段翻译: 成色双向 ===")
print(f" '95新' → {translator.quality_to_int('95新')} (对齐闲鱼stuff_status)")
print(f" '全新未拆封' → {translator.quality_to_int('全新未拆封')}")
print(f" 'abc' → {translator.quality_to_int('abc')} (未知→0)")
uni = Product(sku_id="x", platform="x", title="", price=Money(0),
condition_int=85, condition_text="85新",
quality_desc="轻微磨损", attributes={"容量":"128GB"})
back = translator.from_unified(uni)
print(f" 统一→转转: {back}")跑出来关键几行（正是差异与降级实证）：
=== 转转 MCP: 商品详情 → 统一 Product ===
title=95新 iPhone13 128G price=2300.0 condition_int=95 condition_text=95新
quality_desc=全原装无修 attrs={'容量': '128GB'} market_range={'low': 2000, 'high': 2600}
=== 转转独有: 回收估价 ===
{'sku_id': '', 'platform': 'zhuanzhuan', 'price': 1800.0, 'quality': '90新', ...}
=== 能力不对等: 转转不支持写 ===
⚠️ 转转官方验(C2B2C)库存由平台管控, 不支持商家直接改库存...
⚠️ 转转订单列表未纳入 MCP 开放工具...
=== 能力探测: 按能力选平台 ===
valuation: 推荐平台 = ['zhuanzhuan']
market_price: 推荐平台 = ['zhuanzhuan']
ship: 推荐平台 = []
product_write: 推荐平台 = []
=== 字段翻译: 成色双向 ===
'95新' → 95 (对齐闲鱼stuff_status)
'全新未拆封' → 100
'abc' → 0 (未知→0)
统一→转转: {'title': '', 'price': '0.00', 'quality': '85新', 'attributes': [...]}五、四个接入铁律
能力不对等必须显式建模：
CapabilityProbe+CapabilityNotAvailable异常，让业务代码知道什么时候该降级——转转侧写操作直接抛出（不静默失败），调用方 catch 后回退到闲鱼/人工。成色双向翻译要容错：
quality_to_int对未知字符串返回 0（不抛异常），避免脏数据打断同步；同时保留condition_text/quality_desc原文，防止信息丢失。MCP 凭证 ≠ OAuth：转转用 MCP 凭证 + 工具级 scope，没有"店铺SessionKey"概念——适配层
McpCredential只存tool_scope，不要硬套闲鱼的 token_store。只读为主是设计如此不是缺陷：转转的"可信二手"定位决定了交易闭环在平台侧（官方验、质保、7天无理由），ERP 角色是"辅助决策（行情/估价）+ 入口导流"，不是"控制交易"——别用闲鱼的全链路范式强套转转。
六、与闲鱼适配层的对称设计
commerce-mesh/adapters/ ├── idle/ (闲鱼: TOP OAuth, 全链路交易) │ ├── auth.py ← TokenManager (SessionKey) │ ├── ship.py ← IdleIsvShipClient │ ├── refund.py ← RefundSyncHandler │ ├── publish.py ← IdleItemPublishMapper │ └── verifier.py ← PublishVerifier ├── zhuanzhuan/ (转转: MCP, 只读+估价) │ ├── credential.py ← McpCredential (scope) │ ├── mcp_client.py ← ZhuanzhuanMcpClient │ ├── translator.py ← ZhuanzhuanFieldTranslator │ └── adapter.py ← ZhuanzhuanAdapter (本篇) └── common/ ├── product.py ← Product/Order (统一实体) └── repository.py ← ProductRepository/OrderRepository (端口)
关键：两个 Adapter 实现同一套端口，上层
MarketplaceOrchestrator（前篇）不区分平台——probe().choose_platform("ship") 自动选闲鱼，probe().choose_platform("valuation") 自动选转转。七、和前几篇的衔接
把ZhuanzhuanAdapter作为前篇MarketplaceOrchestrator的第8个平台 Adapter（闲鱼、淘宝、1688、拼多多、抖店、亚马逊、eBay + 转转）：
统一端口：
ProductRepository/OrderRepository与前七家一致，CapabilityProbe挂在调度器入口，调用前按能力选平台；字段翻译：
ZhuanzhuanFieldTranslator复用前篇IdleItemPublishMapper的成色字典（反向映射），三平台（闲鱼/淘宝/1688）的stuff_status与转转quality在master_sku主数据层（前篇StockEngine）汇合；合规：转转 App 私有接口（订单/发货）若未来要走，必须经
ComplianceGate审批（默认禁用采集通道），MCP 官方工具走绿色official_api；行情能力：
market_price()喂前篇的定价/跟价决策，estimate_recycle()喂回收业务——这是转转相对其他七家的差异化增量，别浪费；降级告警：
CapabilityNotAvailable进ObservabilityMiddleware，当转转官方开放写能力时自动升级探测、无需改业务代码。
适配层的价值不是"多接一个平台"，而是"把生态差异（重履约 vs 轻撮合）翻译成统一的领域语言"——上层业务只用关心 Product/Order，不用关心底下是 TOP、MCP 还是私有接口。
要不要我把
zhuanzhuan_adapter.py + ZhuanzhuanFieldTranslator 合进 commerce-mesh/adapters/zhuanzhuan/，并让前篇的 StockEngine（三平台同源库存）升级为"四平台"，把转转的"官方验库存由平台控"作为特殊的只读库存源处理？