Files
lzwcai-mcp/k3cloud_mcp/k3cloud_mcp/client.py
tanjianbin 7e33e398ce feat(k3cloud): 新增单据查看功能并重构演示代码
- 在 client.py 中新增 view() 函数,支持调用金蝶云星空 DynamicFormService.View 接口
- 在 tools.py 中新增 view_form 工具,供 MCP 客户端调用
- 添加演示脚本 demo_view.py 和示例配置文件 demo_view.json
- 重构 demo_save.py 为通用演示脚本,支持自定义 payload 文件
- 删除过时的 demo.json,新增 userful_save.json 作为实用示例
- 添加项目交接文档,详细说明各子项目架构和配置
- 优化 payload 序列化逻辑,提取为共享函数 _serialize_payload
2026-05-08 17:15:20 +08:00

83 lines
2.4 KiB
Python

import json
from typing import Any, Dict
import requests
from .config import Config
from .signing import build_headers
def build_save_url(path: str) -> str:
"""Build the Save API URL."""
return f"{Config.BASE_URL.rstrip('/')}{path}"
def _serialize_payload(payload: Any, field_name: str) -> str:
if payload is None:
raise ValueError(f"{field_name} 不能为空")
if isinstance(payload, str):
payload_value = payload.strip()
if not payload_value:
raise ValueError(f"{field_name} 不能为空字符串")
return payload_value
try:
return json.dumps(payload, ensure_ascii=False)
except TypeError as exc:
raise TypeError(
f"{field_name} 必须是可序列化的 JSON 对象或 JSON 字符串"
) from exc
def save(formid: str, data_payload: Any, timeout: int = 30) -> Dict[str, Any]:
"""
Call the K3Cloud Save API.
`data_payload` can be a JSON object or a pre-serialized JSON string.
"""
save_service_path = "/Kingdee.BOS.WebApi.ServicesStub.DynamicFormService.Save.common.kdsvc"
normalized_formid = str(formid or "").strip()
if not normalized_formid:
raise ValueError("formid 不能为空")
payload_value = _serialize_payload(data_payload, "data_payload")
payload = {
"formid": normalized_formid,
"data": payload_value,
}
response = requests.post(
build_save_url(save_service_path),
json=payload,
headers=build_headers(save_service_path),
timeout=timeout,
)
response.raise_for_status()
return response.json()
def view(formid: str, payload: Any, timeout: int = 30) -> Dict[str, Any]:
"""
Call the K3Cloud View API.
`payload` can be a JSON object or a pre-serialized JSON string.
"""
view_service_path = "/Kingdee.BOS.WebApi.ServicesStub.DynamicFormService.View.common.kdsvc"
normalized_formid = str(formid or "").strip()
if not normalized_formid:
raise ValueError("formid 不能为空")
payload_value = _serialize_payload(payload, "payload")
request_body = {
"formid": normalized_formid,
"data": payload_value,
}
response = requests.post(
build_save_url(view_service_path),
json=request_body,
headers=build_headers(view_service_path),
timeout=timeout,
)
response.raise_for_status()
return response.json()