Coverage for src / npm_mcp / server.py: 98%
188 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-21 06:26 +0400
1"""MCP server for Nginx Proxy Manager."""
3import asyncio
4import json
5from typing import Any
6from mcp.server import Server
7from mcp.server.stdio import stdio_server
8from mcp.types import Tool, TextContent
9from .client import create_client_from_env, NPMClient
10from .models import (
11 ProxyHost, Certificate, AccessList, RedirectionHost,
12 Stream, DeadHost, User, Setting, AuditLogEntry,
13)
16app = Server("nginx-proxy-manager")
17npm_client: NPMClient | None = None
20def _id_schema(name: str, desc: str) -> dict:
21 return {
22 "type": "object",
23 "properties": {name: {"type": "integer", "description": desc}},
24 "required": [name],
25 }
28def _empty_schema() -> dict:
29 return {"type": "object", "properties": {}}
32@app.list_tools()
33async def list_tools() -> list[Tool]:
34 return [
35 # --- Proxy Hosts ---
36 Tool(name="list_proxy_hosts", description="List all proxy hosts configured in NPM", inputSchema=_empty_schema()),
37 Tool(name="get_proxy_host", description="Get details of a specific proxy host by ID", inputSchema=_id_schema("host_id", "The ID of the proxy host")),
38 Tool(
39 name="create_proxy_host",
40 description="Create a new proxy host",
41 inputSchema={
42 "type": "object",
43 "properties": {
44 "domain_names": {"type": "array", "items": {"type": "string"}, "description": "List of domain names"},
45 "forward_scheme": {"type": "string", "enum": ["http", "https"], "default": "http"},
46 "forward_host": {"type": "string", "description": "IP or hostname to forward to"},
47 "forward_port": {"type": "integer", "description": "Port to forward to"},
48 "certificate_id": {"type": "integer", "description": "SSL certificate ID"},
49 "ssl_forced": {"type": "boolean", "default": False},
50 "block_exploits": {"type": "boolean", "default": True},
51 "advanced_config": {"type": "string", "default": ""},
52 },
53 "required": ["domain_names", "forward_host", "forward_port"],
54 },
55 ),
56 Tool(
57 name="update_proxy_host",
58 description="Update an existing proxy host",
59 inputSchema={
60 "type": "object",
61 "properties": {
62 "host_id": {"type": "integer", "description": "The ID of the proxy host to update"},
63 "domain_names": {"type": "array", "items": {"type": "string"}},
64 "forward_scheme": {"type": "string", "enum": ["http", "https"]},
65 "forward_host": {"type": "string"},
66 "forward_port": {"type": "integer"},
67 "certificate_id": {"type": "integer"},
68 "ssl_forced": {"type": "boolean"},
69 "block_exploits": {"type": "boolean"},
70 "advanced_config": {"type": "string"},
71 },
72 "required": ["host_id"],
73 },
74 ),
75 Tool(name="delete_proxy_host", description="Delete a proxy host by ID", inputSchema=_id_schema("host_id", "The ID of the proxy host to delete")),
76 Tool(name="enable_proxy_host", description="Enable a proxy host", inputSchema=_id_schema("host_id", "The ID of the proxy host to enable")),
77 Tool(name="disable_proxy_host", description="Disable a proxy host", inputSchema=_id_schema("host_id", "The ID of the proxy host to disable")),
79 # --- Redirection Hosts ---
80 Tool(name="list_redirection_hosts", description="List all redirection hosts", inputSchema=_empty_schema()),
81 Tool(name="get_redirection_host", description="Get a specific redirection host by ID", inputSchema=_id_schema("host_id", "The ID of the redirection host")),
82 Tool(
83 name="create_redirection_host",
84 description="Create a new redirection host (HTTP redirect)",
85 inputSchema={
86 "type": "object",
87 "properties": {
88 "domain_names": {"type": "array", "items": {"type": "string"}, "description": "List of domain names"},
89 "forward_scheme": {"type": "string", "enum": ["auto", "http", "https"], "default": "auto"},
90 "forward_http_code": {"type": "integer", "description": "HTTP redirect code (301, 302, etc.)", "enum": [300, 301, 302, 303, 304, 305, 307, 308]},
91 "forward_domain_name": {"type": "string", "description": "Domain to redirect to"},
92 "preserve_path": {"type": "boolean", "default": False},
93 "certificate_id": {"type": "integer"},
94 "ssl_forced": {"type": "boolean", "default": False},
95 "block_exploits": {"type": "boolean", "default": True},
96 "advanced_config": {"type": "string", "default": ""},
97 },
98 "required": ["domain_names", "forward_http_code", "forward_domain_name"],
99 },
100 ),
101 Tool(
102 name="update_redirection_host",
103 description="Update an existing redirection host",
104 inputSchema={
105 "type": "object",
106 "properties": {
107 "host_id": {"type": "integer", "description": "The ID of the redirection host to update"},
108 "domain_names": {"type": "array", "items": {"type": "string"}},
109 "forward_scheme": {"type": "string", "enum": ["auto", "http", "https"]},
110 "forward_http_code": {"type": "integer"},
111 "forward_domain_name": {"type": "string"},
112 "preserve_path": {"type": "boolean"},
113 "certificate_id": {"type": "integer"},
114 "ssl_forced": {"type": "boolean"},
115 "block_exploits": {"type": "boolean"},
116 "advanced_config": {"type": "string"},
117 },
118 "required": ["host_id"],
119 },
120 ),
121 Tool(name="delete_redirection_host", description="Delete a redirection host by ID", inputSchema=_id_schema("host_id", "The ID of the redirection host to delete")),
122 Tool(name="enable_redirection_host", description="Enable a redirection host", inputSchema=_id_schema("host_id", "The ID of the redirection host to enable")),
123 Tool(name="disable_redirection_host", description="Disable a redirection host", inputSchema=_id_schema("host_id", "The ID of the redirection host to disable")),
125 # --- Streams ---
126 Tool(name="list_streams", description="List all TCP/UDP stream proxies", inputSchema=_empty_schema()),
127 Tool(name="get_stream", description="Get a specific stream by ID", inputSchema=_id_schema("stream_id", "The ID of the stream")),
128 Tool(
129 name="create_stream",
130 description="Create a new TCP/UDP stream proxy",
131 inputSchema={
132 "type": "object",
133 "properties": {
134 "incoming_port": {"type": "integer", "description": "Port to listen on (1-65535)"},
135 "forwarding_host": {"type": "string", "description": "Host to forward to"},
136 "forwarding_port": {"type": "integer", "description": "Port to forward to (1-65535)"},
137 "tcp_forwarding": {"type": "boolean", "default": True},
138 "udp_forwarding": {"type": "boolean", "default": False},
139 "certificate_id": {"type": "integer"},
140 },
141 "required": ["incoming_port", "forwarding_host", "forwarding_port"],
142 },
143 ),
144 Tool(
145 name="update_stream",
146 description="Update an existing stream",
147 inputSchema={
148 "type": "object",
149 "properties": {
150 "stream_id": {"type": "integer", "description": "The ID of the stream to update"},
151 "incoming_port": {"type": "integer"},
152 "forwarding_host": {"type": "string"},
153 "forwarding_port": {"type": "integer"},
154 "tcp_forwarding": {"type": "boolean"},
155 "udp_forwarding": {"type": "boolean"},
156 "certificate_id": {"type": "integer"},
157 },
158 "required": ["stream_id"],
159 },
160 ),
161 Tool(name="delete_stream", description="Delete a stream by ID", inputSchema=_id_schema("stream_id", "The ID of the stream to delete")),
162 Tool(name="enable_stream", description="Enable a stream", inputSchema=_id_schema("stream_id", "The ID of the stream to enable")),
163 Tool(name="disable_stream", description="Disable a stream", inputSchema=_id_schema("stream_id", "The ID of the stream to disable")),
165 # --- Dead Hosts (404) ---
166 Tool(name="list_dead_hosts", description="List all 404 dead hosts", inputSchema=_empty_schema()),
167 Tool(name="get_dead_host", description="Get a specific dead host by ID", inputSchema=_id_schema("host_id", "The ID of the dead host")),
168 Tool(
169 name="create_dead_host",
170 description="Create a new 404 dead host",
171 inputSchema={
172 "type": "object",
173 "properties": {
174 "domain_names": {"type": "array", "items": {"type": "string"}, "description": "List of domain names"},
175 "certificate_id": {"type": "integer"},
176 "ssl_forced": {"type": "boolean", "default": False},
177 "hsts_enabled": {"type": "boolean", "default": False},
178 "hsts_subdomains": {"type": "boolean", "default": False},
179 "http2_support": {"type": "boolean", "default": False},
180 "advanced_config": {"type": "string", "default": ""},
181 },
182 "required": ["domain_names"],
183 },
184 ),
185 Tool(
186 name="update_dead_host",
187 description="Update an existing dead host",
188 inputSchema={
189 "type": "object",
190 "properties": {
191 "host_id": {"type": "integer", "description": "The ID of the dead host to update"},
192 "domain_names": {"type": "array", "items": {"type": "string"}},
193 "certificate_id": {"type": "integer"},
194 "ssl_forced": {"type": "boolean"},
195 "hsts_enabled": {"type": "boolean"},
196 "hsts_subdomains": {"type": "boolean"},
197 "http2_support": {"type": "boolean"},
198 "advanced_config": {"type": "string"},
199 },
200 "required": ["host_id"],
201 },
202 ),
203 Tool(name="delete_dead_host", description="Delete a dead host by ID", inputSchema=_id_schema("host_id", "The ID of the dead host to delete")),
204 Tool(name="enable_dead_host", description="Enable a dead host", inputSchema=_id_schema("host_id", "The ID of the dead host to enable")),
205 Tool(name="disable_dead_host", description="Disable a dead host", inputSchema=_id_schema("host_id", "The ID of the dead host to disable")),
207 # --- Certificates ---
208 Tool(name="list_certificates", description="List all SSL certificates in NPM", inputSchema=_empty_schema()),
209 Tool(name="get_certificate", description="Get a specific SSL certificate by ID", inputSchema=_id_schema("certificate_id", "The ID of the certificate")),
210 Tool(
211 name="request_certificate",
212 description="Request a new Let's Encrypt SSL certificate",
213 inputSchema={
214 "type": "object",
215 "properties": {
216 "domain_names": {"type": "array", "items": {"type": "string"}, "description": "List of domains for the certificate"},
217 "nice_name": {"type": "string", "description": "Friendly name for the certificate"},
218 },
219 "required": ["domain_names", "nice_name"],
220 },
221 ),
222 Tool(name="delete_certificate", description="Delete an SSL certificate by ID", inputSchema=_id_schema("certificate_id", "The ID of the certificate to delete")),
223 Tool(name="renew_certificate", description="Renew a Let's Encrypt certificate", inputSchema=_id_schema("certificate_id", "The ID of the certificate to renew")),
224 Tool(name="list_dns_providers", description="List supported DNS providers for DNS challenge certificates", inputSchema=_empty_schema()),
225 Tool(
226 name="test_http_challenge",
227 description="Test if domains are reachable for HTTP-01 ACME challenge",
228 inputSchema={
229 "type": "object",
230 "properties": {
231 "domains": {"type": "array", "items": {"type": "string"}, "description": "List of domains to test"},
232 },
233 "required": ["domains"],
234 },
235 ),
237 # --- Access Lists ---
238 Tool(name="list_access_lists", description="List all access lists (basic auth, IP restrictions)", inputSchema=_empty_schema()),
239 Tool(name="get_access_list", description="Get a specific access list by ID", inputSchema=_id_schema("access_list_id", "The ID of the access list")),
240 Tool(
241 name="create_access_list",
242 description="Create a new access list",
243 inputSchema={
244 "type": "object",
245 "properties": {
246 "name": {"type": "string", "description": "Name of the access list"},
247 "satisfy_any": {"type": "boolean", "default": False},
248 "pass_auth": {"type": "boolean", "default": True},
249 },
250 "required": ["name"],
251 },
252 ),
253 Tool(
254 name="update_access_list",
255 description="Update an existing access list",
256 inputSchema={
257 "type": "object",
258 "properties": {
259 "access_list_id": {"type": "integer"},
260 "name": {"type": "string"},
261 "satisfy_any": {"type": "boolean"},
262 "pass_auth": {"type": "boolean"},
263 },
264 "required": ["access_list_id"],
265 },
266 ),
267 Tool(name="delete_access_list", description="Delete an access list by ID", inputSchema=_id_schema("access_list_id", "The ID of the access list to delete")),
269 # --- Users ---
270 Tool(name="list_users", description="List all NPM users", inputSchema=_empty_schema()),
271 Tool(name="get_user", description="Get a specific user by ID", inputSchema=_id_schema("user_id", "The ID of the user")),
272 Tool(
273 name="create_user",
274 description="Create a new NPM user",
275 inputSchema={
276 "type": "object",
277 "properties": {
278 "name": {"type": "string", "description": "Full name"},
279 "nickname": {"type": "string", "default": ""},
280 "email": {"type": "string", "description": "Email address"},
281 "roles": {"type": "array", "items": {"type": "string"}, "description": "Roles (e.g. admin)"},
282 "is_disabled": {"type": "boolean", "default": False},
283 },
284 "required": ["name", "email"],
285 },
286 ),
287 Tool(
288 name="update_user",
289 description="Update an existing NPM user",
290 inputSchema={
291 "type": "object",
292 "properties": {
293 "user_id": {"type": "integer", "description": "The ID of the user to update"},
294 "name": {"type": "string"},
295 "nickname": {"type": "string"},
296 "email": {"type": "string"},
297 "roles": {"type": "array", "items": {"type": "string"}},
298 "is_disabled": {"type": "boolean"},
299 },
300 "required": ["user_id"],
301 },
302 ),
303 Tool(name="delete_user", description="Delete a user by ID", inputSchema=_id_schema("user_id", "The ID of the user to delete")),
305 # --- Settings ---
306 Tool(name="list_settings", description="List all NPM settings", inputSchema=_empty_schema()),
307 Tool(
308 name="get_setting",
309 description="Get a specific setting by ID",
310 inputSchema={
311 "type": "object",
312 "properties": {"setting_id": {"type": "string", "description": "The setting ID (e.g. 'default-site')"}},
313 "required": ["setting_id"],
314 },
315 ),
316 Tool(
317 name="update_setting",
318 description="Update a setting",
319 inputSchema={
320 "type": "object",
321 "properties": {
322 "setting_id": {"type": "string", "description": "The setting ID"},
323 "value": {"type": "string", "description": "Setting value"},
324 "meta": {"type": "object", "description": "Setting metadata"},
325 },
326 "required": ["setting_id"],
327 },
328 ),
330 # --- Audit Log ---
331 Tool(name="list_audit_log", description="List recent audit log entries", inputSchema=_empty_schema()),
333 # --- Reports ---
334 Tool(name="get_host_report", description="Get host count report (proxy, redirection, stream, dead)", inputSchema=_empty_schema()),
335 ]
338def _json_response(data: Any) -> list[TextContent]:
339 return [TextContent(type="text", text=json.dumps(data, indent=2))]
342def _model_response(obj: Any) -> list[TextContent]:
343 return _json_response(obj.model_dump())
346def _list_response(items: list) -> list[TextContent]:
347 return _json_response([item.model_dump() for item in items])
350def _msg_response(msg: str) -> list[TextContent]:
351 return [TextContent(type="text", text=msg)]
354@app.call_tool()
355async def call_tool(name: str, arguments: Any) -> list[TextContent]:
356 global npm_client
357 if npm_client is None: 357 ↛ 358line 357 didn't jump to line 358 because the condition on line 357 was never true
358 npm_client = create_client_from_env()
360 try:
361 # --- Proxy Hosts ---
362 if name == "list_proxy_hosts":
363 return _list_response(await npm_client.list_proxy_hosts())
364 elif name == "get_proxy_host":
365 return _model_response(await npm_client.get_proxy_host(arguments["host_id"]))
366 elif name == "create_proxy_host":
367 return _model_response(await npm_client.create_proxy_host(ProxyHost(**arguments)))
368 elif name == "update_proxy_host":
369 args = dict(arguments)
370 host_id = args.pop("host_id")
371 current = await npm_client.get_proxy_host(host_id)
372 updated_data = current.model_dump()
373 updated_data.update(args)
374 return _model_response(await npm_client.update_proxy_host(host_id, ProxyHost(**updated_data)))
375 elif name == "delete_proxy_host":
376 await npm_client.delete_proxy_host(arguments["host_id"])
377 return _msg_response("Proxy host deleted successfully")
378 elif name == "enable_proxy_host":
379 await npm_client.enable_proxy_host(arguments["host_id"])
380 return _msg_response("Proxy host enabled successfully")
381 elif name == "disable_proxy_host":
382 await npm_client.disable_proxy_host(arguments["host_id"])
383 return _msg_response("Proxy host disabled successfully")
385 # --- Redirection Hosts ---
386 elif name == "list_redirection_hosts":
387 return _list_response(await npm_client.list_redirection_hosts())
388 elif name == "get_redirection_host":
389 return _model_response(await npm_client.get_redirection_host(arguments["host_id"]))
390 elif name == "create_redirection_host":
391 return _model_response(await npm_client.create_redirection_host(RedirectionHost(**arguments)))
392 elif name == "update_redirection_host":
393 args = dict(arguments)
394 host_id = args.pop("host_id")
395 current = await npm_client.get_redirection_host(host_id)
396 updated_data = current.model_dump()
397 updated_data.update(args)
398 return _model_response(await npm_client.update_redirection_host(host_id, RedirectionHost(**updated_data)))
399 elif name == "delete_redirection_host":
400 await npm_client.delete_redirection_host(arguments["host_id"])
401 return _msg_response("Redirection host deleted successfully")
402 elif name == "enable_redirection_host":
403 await npm_client.enable_redirection_host(arguments["host_id"])
404 return _msg_response("Redirection host enabled successfully")
405 elif name == "disable_redirection_host":
406 await npm_client.disable_redirection_host(arguments["host_id"])
407 return _msg_response("Redirection host disabled successfully")
409 # --- Streams ---
410 elif name == "list_streams":
411 return _list_response(await npm_client.list_streams())
412 elif name == "get_stream":
413 return _model_response(await npm_client.get_stream(arguments["stream_id"]))
414 elif name == "create_stream":
415 return _model_response(await npm_client.create_stream(Stream(**arguments)))
416 elif name == "update_stream":
417 args = dict(arguments)
418 stream_id = args.pop("stream_id")
419 current = await npm_client.get_stream(stream_id)
420 updated_data = current.model_dump()
421 updated_data.update(args)
422 return _model_response(await npm_client.update_stream(stream_id, Stream(**updated_data)))
423 elif name == "delete_stream":
424 await npm_client.delete_stream(arguments["stream_id"])
425 return _msg_response("Stream deleted successfully")
426 elif name == "enable_stream":
427 await npm_client.enable_stream(arguments["stream_id"])
428 return _msg_response("Stream enabled successfully")
429 elif name == "disable_stream":
430 await npm_client.disable_stream(arguments["stream_id"])
431 return _msg_response("Stream disabled successfully")
433 # --- Dead Hosts ---
434 elif name == "list_dead_hosts":
435 return _list_response(await npm_client.list_dead_hosts())
436 elif name == "get_dead_host":
437 return _model_response(await npm_client.get_dead_host(arguments["host_id"]))
438 elif name == "create_dead_host":
439 return _model_response(await npm_client.create_dead_host(DeadHost(**arguments)))
440 elif name == "update_dead_host":
441 args = dict(arguments)
442 host_id = args.pop("host_id")
443 current = await npm_client.get_dead_host(host_id)
444 updated_data = current.model_dump()
445 updated_data.update(args)
446 return _model_response(await npm_client.update_dead_host(host_id, DeadHost(**updated_data)))
447 elif name == "delete_dead_host":
448 await npm_client.delete_dead_host(arguments["host_id"])
449 return _msg_response("Dead host deleted successfully")
450 elif name == "enable_dead_host":
451 await npm_client.enable_dead_host(arguments["host_id"])
452 return _msg_response("Dead host enabled successfully")
453 elif name == "disable_dead_host":
454 await npm_client.disable_dead_host(arguments["host_id"])
455 return _msg_response("Dead host disabled successfully")
457 # --- Certificates ---
458 elif name == "list_certificates":
459 return _list_response(await npm_client.list_certificates())
460 elif name == "get_certificate":
461 return _model_response(await npm_client.get_certificate(arguments["certificate_id"]))
462 elif name == "request_certificate":
463 return _model_response(await npm_client.request_certificate(Certificate(**arguments)))
464 elif name == "delete_certificate":
465 await npm_client.delete_certificate(arguments["certificate_id"])
466 return _msg_response("Certificate deleted successfully")
467 elif name == "renew_certificate":
468 return _model_response(await npm_client.renew_certificate(arguments["certificate_id"]))
469 elif name == "list_dns_providers":
470 return _json_response(await npm_client.list_dns_providers())
471 elif name == "test_http_challenge":
472 return _json_response(await npm_client.test_http_challenge(arguments["domains"]))
474 # --- Access Lists ---
475 elif name == "list_access_lists":
476 return _list_response(await npm_client.list_access_lists())
477 elif name == "get_access_list":
478 return _model_response(await npm_client.get_access_list(arguments["access_list_id"]))
479 elif name == "create_access_list":
480 return _model_response(await npm_client.create_access_list(AccessList(**arguments)))
481 elif name == "update_access_list":
482 args = dict(arguments)
483 access_list_id = args.pop("access_list_id")
484 current = await npm_client.get_access_list(access_list_id)
485 updated_data = current.model_dump()
486 updated_data.update(args)
487 return _model_response(await npm_client.update_access_list(access_list_id, AccessList(**updated_data)))
488 elif name == "delete_access_list":
489 await npm_client.delete_access_list(arguments["access_list_id"])
490 return _msg_response("Access list deleted successfully")
492 # --- Users ---
493 elif name == "list_users":
494 return _list_response(await npm_client.list_users())
495 elif name == "get_user":
496 return _model_response(await npm_client.get_user(arguments["user_id"]))
497 elif name == "create_user":
498 return _model_response(await npm_client.create_user(User(**arguments)))
499 elif name == "update_user":
500 args = dict(arguments)
501 user_id = args.pop("user_id")
502 current = await npm_client.get_user(user_id)
503 updated_data = current.model_dump()
504 updated_data.update(args)
505 return _model_response(await npm_client.update_user(user_id, User(**updated_data)))
506 elif name == "delete_user":
507 await npm_client.delete_user(arguments["user_id"])
508 return _msg_response("User deleted successfully")
510 # --- Settings ---
511 elif name == "list_settings":
512 return _list_response(await npm_client.list_settings())
513 elif name == "get_setting":
514 return _model_response(await npm_client.get_setting(arguments["setting_id"]))
515 elif name == "update_setting":
516 args = dict(arguments)
517 setting_id = args.pop("setting_id")
518 current = await npm_client.get_setting(setting_id)
519 updated_data = current.model_dump()
520 updated_data.update(args)
521 return _model_response(await npm_client.update_setting(setting_id, Setting(**updated_data)))
523 # --- Audit Log ---
524 elif name == "list_audit_log":
525 return _list_response(await npm_client.list_audit_log())
527 # --- Reports ---
528 elif name == "get_host_report":
529 return _json_response(await npm_client.get_host_report())
531 else:
532 raise ValueError(f"Unknown tool: {name}")
534 except Exception as e:
535 return [TextContent(type="text", text=f"Error: {str(e)}")]
538async def async_main():
539 async with stdio_server() as (read_stream, write_stream):
540 await app.run(read_stream, write_stream, app.create_initialization_options())
543def main():
544 asyncio.run(async_main())
547if __name__ == "__main__":
548 main()