data-collector-mcp/data_collector_mcp/common_server.py

from __future__ import annotations

from typing import Any

from .auth import load_projects_config
from .collector_api import (
    connect_device as api_connect_device,
    disconnect_device as api_disconnect_device,
    list_device_points as api_list_device_points,
    list_devices as api_list_devices,
)
from .mcp_app import mcp


@mcp.tool(
    name="project.list",
    title="Project List",
    description=(
        "List enabled 汇采 projects available to this MCP service, including "
        "project_key, project_name, base_url, and data_collector_base_url. "
        "Call this first to choose project_key."
    ),
    tags={"project", "list"},
)
def project_list() -> dict[str, Any]:
    projects = load_projects_config()
    result = [
        {
            "project_key": item["project_key"],
            "project_name": item["project_name"],
            "base_url": item["base_url"],
            "data_collector_base_url": item["data_collector_base_url"],
        }
        for item in projects
        if item["enabled"]
    ]
    result.sort(key=lambda item: item["project_key"])
    return {"projects": result, "total": len(result)}


@mcp.tool(
    name="collector.device_list",
    description=(
        "汇采-查询设备列表。调用 {data_collector_base_url}/api/collector/device。"
        "用于查看设备树、设备分组、设备连接状态、设备采集状态和点位数量；"
        "不返回点位明细、点位采集状态或点位当前值。"
        "如果目标是查看某个设备下点位的采集状态和值，请使用 collector.device_points。"
        "num_points 默认 false；只有需要统计设备或点位分组下的点位数量时才传 true。"
        "响应透传上游 JSON，state=0 表示业务成功。"
    ),
)
def collector_device_list(project_key: str, num_points: bool = False) -> dict[str, Any]:
    return api_list_devices(project_key, num_points=num_points)


@mcp.tool(
    name="collector.device_connect",
    description=(
        "汇采-连接设备。调用 "
        "{data_collector_base_url}/api/collector/common/device/set_connect_status，"
        "请求 status=2。用于让指定设备进入已连接状态；连接成功不代表正在采集，"
        "采集状态请看响应 data.running_status 或后续查询设备/点位状态。"
        "device_id 是设备列表中的设备 id；device_type 默认 modbus，其他设备类型可传 "
        "s7、bacnet、ethernet-ip、opc-ua、opc-da、snmp、iec104。"
        "响应透传上游 JSON，state=0 表示业务成功。常见状态："
        "data.status 1=未连接、2=已连接、3=连接异常；"
        "data.running_status 0=未采集、1=采集中、2=采集异常。"
    ),
)
def collector_device_connect(
    project_key: str,
    device_id: int,
    device_type: str = "modbus",
) -> dict[str, Any]:
    return api_connect_device(project_key, device_id=device_id, device_type=device_type)


@mcp.tool(
    name="collector.device_disconnect",
    description=(
        "汇采-断开设备。调用 "
        "{data_collector_base_url}/api/collector/common/device/set_connect_status，"
        "请求 status=1。用于停止指定设备连接/采集相关状态，使设备回到未连接或空闲状态。"
        "device_id 是设备列表中的设备 id；device_type 默认 modbus，其他设备类型可传 "
        "s7、bacnet、ethernet-ip、opc-ua、opc-da、snmp、iec104。"
        "响应透传上游 JSON，state=0 表示业务成功。常见状态："
        "data.status 1=未连接、2=已连接、3=连接异常；"
        "data.running_status 0=未采集、1=采集中、2=采集异常。"
    ),
)
def collector_device_disconnect(
    project_key: str,
    device_id: int,
    device_type: str = "modbus",
) -> dict[str, Any]:
    return api_disconnect_device(project_key, device_id=device_id, device_type=device_type)


@mcp.tool(
    name="collector.device_points",
    description=(
        "汇采-查询设备点位列表。调用 "
        "{data_collector_base_url}/api/collector/common/device/get_collect_point。"
        "主要用于查看某个设备下点位的采集状态和值：Modbus/S7 返回 data.point[].status "
        "和 data.point[].present_value，status 0=未采集、1=采集正常、2=采集异常；"
        "present_value 是当前内存中的点位最新值。group_id 默认 0 表示查询全部点位，"
        "传具体点位分组 id 时只返回该分组下点位。device_type 默认 modbus，其他设备类型可传 "
        "s7、bacnet、ethernet-ip、opc-ua、opc-da、snmp、iec104。"
        "响应透传上游 JSON，state=0 表示业务成功。Modbus 点位常见字段包括 id、point_id、"
        "name、address、type、device_id、status、function_code、present_value、scale_ratio、"
        "value_offset、group_id、bit、update_time；S7 点位常见字段包括 id、point_id、name、"
        "address、data_type、register_type、device_id、status、present_value、scale_ratio、"
        "value_offset、group_id、update_time。"
    ),
)
def collector_device_points(
    project_key: str,
    device_id: int,
    device_type: str = "modbus",
    group_id: int = 0,
) -> dict[str, Any]:
    return api_list_device_points(
        project_key,
        device_id=device_id,
        device_type=device_type,
        group_id=group_id,
    )