112 lines
3.5 KiB
Python
112 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""STRATO DNS MCP for a single configured zone."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
from strato_client import StratoClient, StratoConfig, StratoError
|
|
|
|
|
|
mcp = FastMCP(
|
|
"strato-dns",
|
|
instructions=(
|
|
"Read and manage CNAME records in the configured STRATO DNS zone. "
|
|
"Write tools are available only when STRATO_WRITE_ENABLED=true."
|
|
),
|
|
host="0.0.0.0",
|
|
port=int(os.environ.get("PORT", "8000")),
|
|
stateless_http=True,
|
|
)
|
|
|
|
|
|
def _json(value: object) -> str:
|
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
|
|
|
|
|
|
@mcp.tool()
|
|
def strato_connection_status() -> str:
|
|
"""Log in and verify that the configured DNS zone can be read. Makes no change."""
|
|
try:
|
|
config = StratoConfig.from_env()
|
|
records = StratoClient(config).list_txt_and_cname_records()
|
|
return _json({
|
|
"connected": True,
|
|
"domain": config.domain,
|
|
"record_count": len(records),
|
|
"cname_count": sum(record.type == "CNAME" for record in records),
|
|
"read_only": True,
|
|
"write_enabled": config.write_enabled,
|
|
})
|
|
except StratoError as exc:
|
|
return _json({"connected": False, "error": str(exc), "read_only": True})
|
|
|
|
|
|
@mcp.tool()
|
|
def strato_list_cnames() -> str:
|
|
"""List CNAME records for the one configured STRATO zone. Makes no change."""
|
|
try:
|
|
config = StratoConfig.from_env()
|
|
records = StratoClient(config).list_cnames()
|
|
return _json({
|
|
"domain": config.domain,
|
|
"count": len(records),
|
|
"records": [record.as_dict() for record in records[:200]],
|
|
"truncated": len(records) > 200,
|
|
"read_only": True,
|
|
"write_enabled": config.write_enabled,
|
|
})
|
|
except StratoError as exc:
|
|
return _json({"error": str(exc), "read_only": True})
|
|
|
|
|
|
@mcp.tool()
|
|
def strato_create_cname(prefix: str, target: str) -> str:
|
|
"""Create one CNAME. Prefix may be relative or end in the configured zone."""
|
|
try:
|
|
config = StratoConfig.from_env()
|
|
record = StratoClient(config).create_cname(prefix, target)
|
|
return _json({"created": True, "domain": config.domain, "record": record.as_dict()})
|
|
except StratoError as exc:
|
|
return _json({"created": False, "error": str(exc)})
|
|
|
|
|
|
@mcp.tool()
|
|
def strato_update_cname(prefix: str, target: str, new_prefix: str | None = None) -> str:
|
|
"""Change a CNAME target and optionally rename its prefix."""
|
|
try:
|
|
config = StratoConfig.from_env()
|
|
before, after = StratoClient(config).update_cname(
|
|
prefix, target, new_prefix=new_prefix
|
|
)
|
|
return _json({
|
|
"updated": True,
|
|
"domain": config.domain,
|
|
"before": before.as_dict(),
|
|
"after": after.as_dict(),
|
|
})
|
|
except StratoError as exc:
|
|
return _json({"updated": False, "error": str(exc)})
|
|
|
|
|
|
@mcp.tool()
|
|
def strato_delete_cname(prefix: str) -> str:
|
|
"""Delete exactly one existing CNAME identified by its prefix."""
|
|
try:
|
|
config = StratoConfig.from_env()
|
|
deleted = StratoClient(config).delete_cname(prefix)
|
|
return _json({
|
|
"deleted": True,
|
|
"domain": config.domain,
|
|
"record": deleted.as_dict(),
|
|
})
|
|
except StratoError as exc:
|
|
return _json({"deleted": False, "error": str(exc)})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run(transport=os.environ.get("MCP_TRANSPORT", "streamable-http"))
|