"""
Device Log MCP Server
---------------------
A simple MCP server exposing simulated device logs for different services.
Supports JSON testing via curl without session requirements.
"""
import random
import logging
from mcp.server.fastmcp import FastMCP
# ==============================
# Logging setup
# ==============================
# Suppress noisy internal streamable HTTP errors
logging.getLogger("mcp.server.streamable_http").setLevel(logging.CRITICAL)
# ==============================
# Server initialization
# ==============================
# Stateless + JSON mode for easy local testing
mcp = FastMCP(
name="Device Log Server",
stateless_http=True, # No session ID required
json_response=True # Respond in plain JSON instead of SSE
)
# ==============================
# Helper function
# ==============================
def fake_log(service: str, serial_number: str) -> str:
"""Generate a fake log line for the given service and serial number."""
samples = {
"activation-service": f"[{serial_number}] Activation completed at step 4",
"lan-service": f"[{serial_number}] LAN connected, IP=10.0.{random.randint(10,99)}.{random.randint(1,255)}",
"wlan-service": f"[{serial_number}] WLAN policy push failed due to cert mismatch",
"kafka": f"[{serial_number}] status=policy_error at 14:23:07"
}
result = samples.get(service, f"No logs for {serial_number} in {service}")
print(f"[LOG] Generated log for {service}: {result}")
return result
# ==============================
# Tools
# ==============================
@mcp.tool()
def get_activation_logs(serial_number: str) -> str:
"""Fetch activation logs for the given device serial number."""
print(f"[CALL] get_activation_logs called with serial_number={serial_number}")
result = fake_log("activation-service", serial_number)
print(f"[RETURN] get_activation_logs → {result}")
return result
@mcp.tool()
def get_lan_logs(serial_number: str) -> str:
"""Fetch LAN logs for the given device serial number."""
print(f"[CALL] get_lan_logs called with serial_number={serial_number}")
result = fake_log("lan-service", serial_number)
print(f"[RETURN] get_lan_logs → {result}")
return result
@mcp.tool()
def get_wlan_logs(serial_number: str) -> str:
"""Fetch WLAN logs for the given device serial number."""
print(f"[CALL] get_wlan_logs called with serial_number={serial_number}")
result = fake_log("wlan-service", serial_number)
print(f"[RETURN] get_wlan_logs → {result}")
return result
@mcp.tool()
def get_kafka_status(serial_number: str) -> str:
"""Fetch device status message from Kafka topic."""
print(f"[CALL] get_kafka_status called with serial_number={serial_number}")
result = fake_log("kafka", serial_number)
print(f"[RETURN] get_kafka_status → {result}")
return result
# ==============================
# Entry point
# ==============================
def main():
"""Run the MCP server with HTTP transport on port 4000."""
mcp.settings.port = 4000
print("🚀 Starting Device Log MCP Server on port 4000...")
mcp.run(transport="streamable-http")
if __name__ == "__main__":
main()
