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=(
        "汇采-返回目前所有设备及其详情信息。返回内容中的设备字段包括不限于 name 设备名、"
        "type 设备类型、num_points 点位数量、running_status 运行状态："
        "0 已停止，1 正常，2 异常。type 为 devicegroup 时表示设备分组，"
        "需要继续检查 groups 下的设备。"
        "num_points 默认 false；只有需要统计设备或点位分组下的点位数量时才传 true。"
    ),
)
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.running_status"
        "device_id 是设备列表中的设备 id；device_type 可传modbus、s7、bacnet、ethernet-ip、opc-ua、opc-da、snmp、iec104。"
        "响应data字段下常见状态："
        "data.status 1=未连接、2=已连接、3=连接异常；"
        "data.running_status 0=未采集、1=采集中、2=采集异常。"
    ),
)
def collector_device_connect(
    project_key: str,
    device_id: int,
    device_type: str,
) -> dict[str, Any]:
    return api_connect_device(project_key, device_id=device_id, device_type=device_type)


@mcp.tool(
    name="collector.device_disconnect",
    description=(
        "汇采-断开设备。用于停止指定设备连接/采集相关状态，使设备回到未连接或空闲状态"
        "device_id 是设备列表中的设备 id；device_type 可传modbus、s7、bacnet、ethernet-ip、opc-ua、opc-da、snmp、iec104。"
        "响应data字段下常见状态："
        "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=(
        "汇采-查询设备下的所有点位。device_id 是设备id, device_type 可传modbus、s7、bacnet、ethernet-ip、opc-ua、opc-da、snmp、iec104。"
        "status 0=未采集、1=采集正常、2=采集异常, present_value 是当前值"
    ),
)
def collector_device_points(
    project_key: str,
    device_id: int,
    device_type: str,
) -> dict[str, Any]:
    return api_list_device_points(
        project_key,
        device_id=device_id,
        device_type=device_type,
    )