diff --git a/account_resource/init.py b/account_resource/init.py
index 9b97979..17fd295 100644
--- a/account_resource/init.py
+++ b/account_resource/init.py
@@ -320,6 +320,26 @@ def load_account_resource():
         env._resource_pricing_resolvers = existing
         logger.info('产品层解析器注册表未就绪，已暂存 resolver: %s', RESOURCE_TYPE)
 
+    # 注册 product_management 资源模块接口（env.product_interface）
+    from .product_interface import (
+        get_product_display, check_product_availability, check_product_consumable,
+        execute_product_service, execute_product_service_stream, calculate_product_cost)
+    env.product_interface = {
+        'module_name': MODULE_NAME,
+        'get_product_display': get_product_display,
+        'check_product_availability': check_product_availability,
+        'check_product_consumable': check_product_consumable,
+        'execute_product_service': execute_product_service,
+        'execute_product_service_stream': execute_product_service_stream,
+        'calculate_product_cost': calculate_product_cost,
+        'load_product_category_product': load_product_category_product,
+    }
+
     logger.info('account_resource module loaded (v%s, resource_type=%s)',
                 MODULE_VERSION, RESOURCE_TYPE)
     return True
+
+
+# 产品类别/产品导入接口 —— 供 product_management 经 importlib.import_module 访问。
+# 延迟到所有函数定义之后再 import，避免与 product_interface 的循环导入。
+from .product_interface import load_product_category_product  # noqa: E402
diff --git a/account_resource/product_interface.py b/account_resource/product_interface.py
new file mode 100644
index 0000000..38432ab
--- /dev/null
+++ b/account_resource/product_interface.py
@@ -0,0 +1,176 @@
+"""account_resource 产品接口 — 供 product_management 调用。
+
+标准资源模块接口（参照 llmage/product_interface.py）：
+    resource_ref_id = acctres_spec.id（产品表的 resource_ref_id 存这个）
+
+product_management 通过 env.product_interface 拿到这些函数，
+实现「资源→产品→定价」的桥接。定价按 spec × charge_mode 映射到 ppid。
+"""
+from ahserver.serverenv import ServerEnv
+
+from .init import (
+    get_account_spec,
+    acctres_pricing_map_list,
+    get_account_ppid,
+    get_account_privileges,
+)
+
+
+async def _resolve_spec(resource_ref_id):
+    """resource_ref_id (= acctres_spec.id) → spec 记录。"""
+    spec = await get_account_spec(resource_ref_id)
+    if not spec:
+        return None, '账号规格 %s 不存在' % resource_ref_id
+    return spec, None
+
+
+async def load_product_category_product(parent_category_id):
+    """返回账号类别 + 账号产品（试用/月租/年租），供 product_management 导入。
+
+    Returns:
+        {'success', 'categories': [{source_id,name,description,product_type,
+          product_type_title,sort_order}],
+         'products': [{source_category_id,resource_ref_id,product_code,
+          product_name,product_type,brief_intro,sort_order,providerid}]}
+    """
+    specs = await acctres_spec_list({'status': 'active'})
+    if not specs:
+        return {'success': False, 'error': '账号资源模块中没有启用规格'}
+
+    categories = [{
+        'source_id': 'account_specs',
+        'name': '账号服务',
+        'description': '账号订阅（试用 / 月租 / 年租）',
+        'product_type': 'account',
+        'product_type_title': '账号',
+        'sort_order': 0,
+    }]
+
+    products = []
+    for s in specs:
+        products.append({
+            'source_category_id': 'account_specs',
+            'resource_ref_id': getattr(s, 'id'),
+            'product_code': getattr(s, 'spec_code'),
+            'product_name': getattr(s, 'spec_name'),
+            'product_type': 'account',
+            'brief_intro': getattr(s, 'description', '') or '',
+            'sort_order': int(getattr(s, 'sort_order', 0) or 0),
+        })
+
+    return {'success': True, 'categories': categories, 'products': products}
+
+
+async def get_product_display(resource_ref_id):
+    """账号规格定价展示（遍历所有计费方式的定价）。"""
+    spec, err = await _resolve_spec(resource_ref_id)
+    if err:
+        return {'success': False, 'message': err}
+
+    env = ServerEnv()
+    maps = await acctres_pricing_map_list({'spec_id': getattr(spec, 'id')})
+    pricing_texts = []
+    for m in maps:
+        ppid = getattr(m, 'ppid', '') or ''
+        if not ppid:
+            continue
+        try:
+            pd = await env.get_pricing_display(ppid)
+            if pd:
+                txt = pd.get('display_text', '') if isinstance(pd, dict) else str(pd)
+                pricing_texts.append('%s: %s' % (getattr(m, 'charge_mode', ''), txt))
+        except Exception:
+            pricing_texts.append('%s: 定价暂不可用' % getattr(m, 'charge_mode', ''))
+
+    return {
+        'success': True,
+        'pricing_text': ' | '.join(pricing_texts) if pricing_texts else '未配置定价',
+        'pricing_detail': {},
+        'extra_info': {
+            'spec_code': getattr(spec, 'spec_code', ''),
+            'spec_name': getattr(spec, 'spec_name', ''),
+        },
+    }
+
+
+async def check_product_availability(resource_ref_id, user_org_id=None):
+    """账号规格可用性检查。"""
+    spec, err = await _resolve_spec(resource_ref_id)
+    if err:
+        return {'available': False, 'reason': err}
+    if getattr(spec, 'status', '') != 'active':
+        return {'available': False, 'reason': '规格已下线'}
+    return {'available': True, 'reason': ''}
+
+
+async def check_product_consumable(resource_ref_id, user_id, user_org_id):
+    """账号开通前预检（余额、重复订阅等）。"""
+    spec, err = await _resolve_spec(resource_ref_id)
+    if err:
+        return {'consumable': False, 'reason': err,
+                'min_balance': 0, 'pricing_available': False}
+    # 定价可用性：至少一个计费方式有 ppid
+    maps = await acctres_pricing_map_list({'spec_id': getattr(spec, 'id')})
+    pricing_available = any(getattr(m, 'ppid', '') for m in maps)
+    return {'consumable': True, 'reason': '',
+            'min_balance': 0, 'pricing_available': pricing_available}
+
+
+async def execute_product_service(resource_ref_id, user_id, user_org_id, request_data):
+    """账号开通服务。实际权益生效（工作空间数/容量/GPU 限制）后续接入，
+    此处按 request_data 里的 charge_mode 完成订购计费。"""
+    spec, err = await _resolve_spec(resource_ref_id)
+    if err:
+        return {'success': False, 'message': err, 'status': 'FAILED'}
+
+    charge_mode = (request_data or {}).get('charge_mode', '')
+    if not charge_mode:
+        return {'success': False, 'message': '缺少 charge_mode', 'status': 'FAILED'}
+
+    # 计费 + 落用量（account_charging 内部解析 ppid 并写 acctres_usage）
+    from .init import account_charging
+    rec = await account_charging(
+        getattr(spec, 'spec_code'), charge_mode, user_id, user_org_id,
+        subscription_id=(request_data or {}).get('subscription_id', ''),
+        quantity=(request_data or {}).get('quantity', 1))
+    return {'success': True,
+            'message': '账号开通成功（权益生效逻辑待接入）',
+            'status': 'SUCCESS',
+            'usage': rec}
+
+
+async def execute_product_service_stream(resource_ref_id, user_id, user_org_id, request_data):
+    """账号开通无流式场景，返回单条结果。"""
+    r = await execute_product_service(resource_ref_id, user_id, user_org_id, request_data)
+    async def _gen():
+        yield {'chunk': r, 'usage_data': None, 'done': True, 'error': None}
+    return _gen()
+
+
+async def calculate_product_cost(resource_ref_id, usage_data, user_org_id=None):
+    """计算账号费用。usage_data 含 charge_mode（trial/month/year）。
+
+    Returns:
+        {'success', 'amount', 'original_amount', 'cost', 'discount', 'pricing_program_id'}
+    """
+    spec, err = await _resolve_spec(resource_ref_id)
+    if err:
+        return {'success': False, 'message': err}
+
+    usage_data = usage_data or {}
+    charge_mode = usage_data.get('charge_mode', '')
+    ppid = await get_account_ppid(getattr(spec, 'id'), charge_mode)
+    if not ppid:
+        return {'success': False, 'message': '无定价方案(ppid为空)'}
+
+    env = ServerEnv()
+    try:
+        prices = await env.buffered_charging(ppid, usage_data)
+        if prices is None:
+            return {'success': False, 'message': '定价计算返回空'}
+    except Exception as e:
+        return {'success': False, 'message': '定价计算失败: %s' % e}
+
+    amount = sum(getattr(p, 'amount', 0) for p in prices)
+    return {'success': True, 'amount': amount, 'original_amount': amount,
+            'cost': 0.0, 'discount': 1.0, 'pricing_program_id': ppid}