Coverage for oracle / oci_fsdr_mcp_server / server.py: 96.18%
137 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-08-03 16:45 +0000
1"""
2Copyright (c) 2025, Oracle and/or its affiliates.
3Licensed under the Universal Permissive License v1.0 as shown at
4https://oss.oracle.com/licenses/upl.
6OCI Full Stack Disaster Recovery MCP Server
8Exposes OCI FSDR operations as MCP tools for MCP-aware clients.
10The server is intentionally schema-agnostic: read operations are thin wrappers
11over paginated list/get APIs, and all write operations go through a single
12generic passthrough (`fsdr_raw_call`). Polymorphic SDK models (members,
13execution options, log locations) are resolved at call time via a `_type`
14discriminator in the payload, so new member types or plan execution types
15never require server changes.
17Environment variables:
18 OCI_AUTH_TYPE : "api_key" (default) or "security_token"
19 OCI_CONFIG_FILE : path to OCI config file (default: ~/.oci/config)
20 FSDR_PROFILE_1 : profile for region 1 (default: FSDR_REGION1)
21 FSDR_PROFILE_2 : profile for region 2 (default: FSDR_REGION2)
22"""
24from __future__ import annotations
26import logging
27import sys
28import uuid
29from functools import wraps
30from pathlib import Path
31from typing import Any, Callable, Dict, Optional
33import oci
34from oci.util import to_dict
35from fastmcp import FastMCP
36from pydantic import Field
38from .auth import get_dr_client
39from .consts import ALLOWED_FSDR_OPERATIONS, DEFAULT_PROFILE_1, DEFAULT_PROFILE_2
40from .models import (
41 DrPlanExecutionLifecycleState,
42 DrPlanType,
43 DrProtectionGroupLifecycleState,
44 FsdrOperation,
45 ListResult,
46 OciResponseResult,
47 WriteResult,
48)
50# ---------------------------------------------------------------------------
51# Logging (stderr only -- never stdout, which would corrupt the MCP stream)
52# ---------------------------------------------------------------------------
54logging.basicConfig(
55 stream=sys.stderr,
56 level=logging.INFO,
57 format="%(asctime)s [%(levelname)s] %(name)s %(message)s",
58)
59log = logging.getLogger("oci_fsdr_mcp")
61# ---------------------------------------------------------------------------
62# MCP server
63# ---------------------------------------------------------------------------
65_PROMPTS_DIR = Path(__file__).parent / "data" / "prompts"
66_PROFILE_DESCRIPTION = (
67 "OCI config profile for the target FSDR region. "
68 f"Defaults to '{DEFAULT_PROFILE_1}'. Common profiles are "
69 f"'{DEFAULT_PROFILE_1}' for region 1 and '{DEFAULT_PROFILE_2}' for region 2."
70)
72mcp = FastMCP(
73 name="oci-fsdr-mcp-server",
74 instructions=(
75 "Tools for OCI Full Stack Disaster Recovery (FSDR). "
76 f"Two region profiles are available: '{DEFAULT_PROFILE_1}' (region 1) and "
77 f"'{DEFAULT_PROFILE_2}' (region 2). "
78 "All tools accept a 'profile' parameter to target the desired region -- "
79 f"defaults to '{DEFAULT_PROFILE_1}' when omitted. "
80 "Read tools (list_*/get_*) return data directly. "
81 "All mutations go through `fsdr_raw_call`, which invokes any "
82 "DisasterRecoveryClient method by name. For polymorphic payloads "
83 "(DRPG members, execution options, log locations), include a `_type` "
84 "key naming the exact SDK model class -- e.g. "
85 "`_type`: 'UpdateDrProtectionGroupMemberDatabaseDetails'. "
86 "Use the built-in prompts for guided workflows: "
87 "check_dr_status, setup_drpg_pair, run_switchover, run_drill, "
88 "run_failover, plan_refresh_workflow, add_members."
89 ),
90)
92# ---------------------------------------------------------------------------
93# Request tracking decorator
94# ---------------------------------------------------------------------------
97def _tool_logger(func: Callable) -> Callable:
98 """Log each tool call with a short request ID for correlation."""
99 @wraps(func)
100 def wrapper(*args: Any, **kwargs: Any) -> Any:
101 req_id = uuid.uuid4().hex[:8]
102 log_context: Dict[str, Any] = {
103 "operation": kwargs.get("operation"),
104 "profile": kwargs.get("profile"),
105 "kwargs": sorted(kwargs),
106 }
107 if isinstance(kwargs.get("parameters"), dict):
108 log_context["parameter_keys"] = sorted(kwargs["parameters"])
109 log.info("[%s] %s called context=%s", req_id, func.__name__, log_context)
110 try:
111 result = func(*args, **kwargs)
112 log.info("[%s] %s succeeded", req_id, func.__name__)
113 return result
114 except Exception as exc:
115 log.error("[%s] %s failed: %s", req_id, func.__name__, exc)
116 raise
117 return wrapper
119# ---------------------------------------------------------------------------
120# Helpers
121# ---------------------------------------------------------------------------
124def _response_to_result(response: oci.response.Response) -> OciResponseResult:
125 data = response.data
126 return OciResponseResult(
127 data=to_dict(data) if data is not None else None,
128 status=response.status,
129 headers=dict(response.headers or {}),
130 )
133def _resolve_model(value: Any) -> Any:
134 """Recursively turn dicts carrying `_type` into OCI SDK model instances.
136 A dict with a `_type: "SomeClassName"` entry is instantiated as
137 `oci.disaster_recovery.models.SomeClassName(**rest)`. Dicts without
138 `_type` and all other values are returned unchanged (but still recursed
139 into). This lets callers describe polymorphic payloads -- e.g. DRPG
140 members, execution options -- without the server needing to know any
141 specific type.
142 """
143 if isinstance(value, dict):
144 if "_type" in value:
145 rest = {k: v for k, v in value.items() if k != "_type"}
146 type_name = value["_type"]
147 model_cls = getattr(oci.disaster_recovery.models, type_name, None)
148 if model_cls is None: 148 ↛ 149line 148 didn't jump to line 149 because the condition on line 148 was never true
149 raise ValueError(
150 f"Unknown oci.disaster_recovery.models class: '{type_name}'"
151 )
152 return model_cls(**{k: _resolve_model(v) for k, v in rest.items()})
153 return {k: _resolve_model(v) for k, v in value.items()}
154 if isinstance(value, list):
155 return [_resolve_model(v) for v in value]
156 return value
158# ---------------------------------------------------------------------------
159# Guided workflow prompts
160# ---------------------------------------------------------------------------
163def _load_prompt(filename: str) -> str:
164 return (_PROMPTS_DIR / filename).read_text(encoding="utf-8")
167@mcp.prompt()
168def setup_drpg_pair() -> str:
169 """Step-by-step guide to create and associate a PRIMARY/STANDBY DRPG pair across two regions."""
170 return _load_prompt("setup_drpg_pair.md")
173@mcp.prompt()
174def check_dr_status() -> str:
175 """Read-only guide to inspect DRPG state, plan health, execution history, and work requests."""
176 return _load_prompt("check_dr_status.md")
179@mcp.prompt()
180def run_switchover() -> str:
181 """Guide for a planned Switchover — reverses PRIMARY/STANDBY roles; both regions must be up."""
182 return _load_prompt("run_switchover.md")
185@mcp.prompt()
186def run_drill() -> str:
187 """Guide to run a DR Drill (START_DRILL then STOP_DRILL) — roles do NOT change, test only."""
188 return _load_prompt("run_drill.md")
191@mcp.prompt()
192def run_failover() -> str:
193 """Guide for an emergency Failover when primary is down, including mandatory post-failover reset."""
194 return _load_prompt("run_failover.md")
197@mcp.prompt()
198def plan_refresh_workflow() -> str:
199 """Guide to refresh DR plans after adding or removing DRPG members."""
200 return _load_prompt("plan_refresh_workflow.md")
203@mcp.prompt()
204def add_members() -> str:
205 """Guide to add compute, database, and other resources to a DRPG."""
206 return _load_prompt("add_members.md")
208# ---------------------------------------------------------------------------
209# Read tools
210# ---------------------------------------------------------------------------
213@mcp.tool()
214@_tool_logger
215def list_dr_protection_groups(
216 compartment_id: str = Field(
217 ..., description="The OCID of the compartment containing DR Protection Groups."
218 ),
219 lifecycle_state: Optional[DrProtectionGroupLifecycleState] = Field(
220 None, description="Optional DR Protection Group lifecycle state filter."
221 ),
222 limit: int = Field(
223 50, description="Maximum number of DR Protection Groups to return.", ge=1, le=1000
224 ),
225 page: Optional[str] = Field(
226 None, description="Opaque opc-next-page token returned by a previous list call."
227 ),
228 profile: str = Field(DEFAULT_PROFILE_1, description=_PROFILE_DESCRIPTION),
229) -> ListResult:
230 """
231 List DR Protection Groups (DRPGs) in a compartment.
233 Args:
234 compartment_id: OCID of the compartment.
235 lifecycle_state: Optional filter e.g. "ACTIVE". If None, all states returned.
236 limit: Max items per page (default 50).
237 page: opc-next-page token for pagination.
238 profile: OCI config profile for the target region. Defaults to FSDR_REGION1.
240 Returns:
241 { "items": [...], "opc_next_page": "...", "total_items": N }
242 """
243 resp = get_dr_client(profile).list_dr_protection_groups(
244 compartment_id=compartment_id,
245 lifecycle_state=lifecycle_state,
246 limit=limit,
247 page=page,
248 )
249 items = [to_dict(pg) for pg in resp.data.items]
250 return ListResult.from_items(items, resp.headers.get("opc-next-page"))
253@mcp.tool()
254@_tool_logger
255def list_dr_plans_for_protection_group(
256 dr_protection_group_id: str = Field(
257 ..., description="The OCID of the DR Protection Group that owns the plans."
258 ),
259 display_name: Optional[str] = Field(
260 None, description="Optional display name filter for DR Plans."
261 ),
262 plan_type: Optional[DrPlanType] = Field(
263 None,
264 description="Optional DR Plan type filter, such as SWITCHOVER or FAILOVER.",
265 ),
266 limit: int = Field(
267 50, description="Maximum number of DR Plans to return.", ge=1, le=1000
268 ),
269 page: Optional[str] = Field(
270 None, description="Opaque opc-next-page token returned by a previous list call."
271 ),
272 profile: str = Field(DEFAULT_PROFILE_1, description=_PROFILE_DESCRIPTION),
273) -> ListResult:
274 """
275 List DR Plans for a DR Protection Group.
277 Args:
278 dr_protection_group_id: OCID of the DR Protection Group.
279 display_name: Optional filter by display name.
280 plan_type: Optional filter -- "SWITCHOVER", "FAILOVER", "START_DRILL", "STOP_DRILL".
281 limit: Max items per page (default 50).
282 page: opc-next-page token for pagination.
283 profile: OCI config profile for the target region.
285 Returns:
286 { "items": [...], "opc_next_page": "...", "total_items": N }
287 """
288 resp = get_dr_client(profile).list_dr_plans(
289 dr_protection_group_id=dr_protection_group_id,
290 display_name=display_name,
291 dr_plan_type=plan_type,
292 limit=limit,
293 page=page,
294 )
295 items = [to_dict(p) for p in resp.data.items]
296 return ListResult.from_items(items, resp.headers.get("opc-next-page"))
299@mcp.tool()
300@_tool_logger
301def list_dr_plan_executions_for_protection_group(
302 dr_protection_group_id: str = Field(
303 ...,
304 description="The OCID of the DR Protection Group that owns the plan executions.",
305 ),
306 lifecycle_state: Optional[DrPlanExecutionLifecycleState] = Field(
307 None, description="Optional DR Plan Execution lifecycle state filter."
308 ),
309 limit: int = Field(
310 50, description="Maximum number of DR Plan Executions to return.", ge=1, le=1000
311 ),
312 page: Optional[str] = Field(
313 None, description="Opaque opc-next-page token returned by a previous list call."
314 ),
315 profile: str = Field(DEFAULT_PROFILE_1, description=_PROFILE_DESCRIPTION),
316) -> ListResult:
317 """
318 List DR Plan Executions for a DR Protection Group.
320 Args:
321 dr_protection_group_id: OCID of the DR Protection Group.
322 lifecycle_state: Optional lifecycle state filter.
323 limit: Max items per page (default 50).
324 page: opc-next-page token for pagination.
325 profile: OCI config profile for the target region.
327 Returns:
328 { "items": [...], "opc_next_page": "...", "total_items": N }
329 """
330 resp = get_dr_client(profile).list_dr_plan_executions(
331 dr_protection_group_id=dr_protection_group_id,
332 lifecycle_state=lifecycle_state,
333 limit=limit,
334 page=page,
335 )
336 items = [to_dict(e) for e in resp.data.items]
337 return ListResult.from_items(items, resp.headers.get("opc-next-page"))
340@mcp.tool()
341@_tool_logger
342def get_dr_protection_group(
343 dr_protection_group_id: str = Field(
344 ..., description="The OCID of the DR Protection Group to retrieve."
345 ),
346 profile: str = Field(DEFAULT_PROFILE_1, description=_PROFILE_DESCRIPTION),
347) -> OciResponseResult:
348 """
349 Get details of a DR Protection Group.
351 Args:
352 dr_protection_group_id: OCID of the DR Protection Group.
353 profile: OCI config profile for the target region.
354 """
355 resp = get_dr_client(profile).get_dr_protection_group(dr_protection_group_id)
356 return _response_to_result(resp)
359@mcp.tool()
360@_tool_logger
361def get_dr_plan(
362 dr_plan_id: str = Field(..., description="The OCID of the DR Plan to retrieve."),
363 profile: str = Field(DEFAULT_PROFILE_1, description=_PROFILE_DESCRIPTION),
364) -> OciResponseResult:
365 """
366 Get details of a DR Plan.
368 Args:
369 dr_plan_id: OCID of the DR Plan.
370 profile: OCI config profile for the target region.
371 """
372 resp = get_dr_client(profile).get_dr_plan(dr_plan_id)
373 return _response_to_result(resp)
376@mcp.tool()
377@_tool_logger
378def get_dr_plan_execution(
379 dr_plan_execution_id: str = Field(
380 ..., description="The OCID of the DR Plan Execution to retrieve."
381 ),
382 profile: str = Field(DEFAULT_PROFILE_1, description=_PROFILE_DESCRIPTION),
383) -> OciResponseResult:
384 """
385 Get details of a DR Plan Execution.
387 Args:
388 dr_plan_execution_id: OCID of the DR Plan Execution.
389 profile: OCI config profile for the target region.
390 """
391 resp = get_dr_client(profile).get_dr_plan_execution(dr_plan_execution_id)
392 return _response_to_result(resp)
395@mcp.tool()
396@_tool_logger
397def get_work_request(
398 work_request_id: str = Field(
399 ..., description="The OCID of the OCI work request to retrieve."
400 ),
401 profile: str = Field(DEFAULT_PROFILE_1, description=_PROFILE_DESCRIPTION),
402) -> OciResponseResult:
403 """
404 Get details of a Work Request (tracks async operations).
406 Args:
407 work_request_id: OCID of the Work Request.
408 profile: OCI config profile for the target region.
409 """
410 resp = get_dr_client(profile).get_work_request(work_request_id)
411 return _response_to_result(resp)
413@mcp.tool()
414@_tool_logger
415def fsdr_raw_call(
416 operation: FsdrOperation = Field(
417 ...,
418 description=(
419 "Allowed DisasterRecoveryClient method name to invoke. Use read tools "
420 "first, then call this for approved FSDR changes."
421 ),
422 ),
423 parameters: Dict[str, Any] = Field(
424 ...,
425 description=(
426 "Keyword arguments for the OCI SDK operation. Use *_details dicts for "
427 "request bodies and `_type` for polymorphic nested OCI SDK models."
428 ),
429 ),
430 profile: str = Field(DEFAULT_PROFILE_1, description=_PROFILE_DESCRIPTION),
431) -> WriteResult:
432 """
433 Invoke any OCI Full Stack Disaster Recovery API by DisasterRecoveryClient
434 method name. This is the single entry point for all mutations -- CREATE,
435 UPDATE, DELETE, associate, execute, refresh, cancel, etc.
437 WARNING: Live, potentially irreversible operations. The caller is
438 responsible for previewing parameters with the user before invoking
439 write operations -- this tool does not implement a dry-run mode.
441 Parameter construction rules:
443 1. Any `*_details` parameter whose value is a dict is converted to the
444 matching SDK model class by naming convention -- e.g.
445 `create_dr_protection_group_details` -> CreateDrProtectionGroupDetails.
447 2. For polymorphic nested payloads (DRPG members, execution options,
448 log locations, etc.) include a `_type` key naming the exact SDK
449 model class. The tool recurses into dicts and lists and instantiates
450 any dict with a `_type` key as that class. Examples:
452 "members": [
453 {"_type": "UpdateDrProtectionGroupMemberDatabaseDetails",
454 "member_id": "ocid1.database..."},
455 {"_type": "UpdateDrProtectionGroupMemberComputeInstanceMovableDetails",
456 "member_id": "ocid1.instance..."}
457 ]
459 "execution_options": {
460 "_type": "SwitchoverExecutionOptionDetails",
461 "are_prechecks_enabled": true,
462 "are_warnings_ignored": false
463 }
465 "log_location": {
466 "_type": "CreateObjectStorageLogLocationDetails",
467 "bucket": "dr-logs",
468 "namespace": "mytenancy"
469 }
471 Refer to the `oci.disaster_recovery.models` module for the full set
472 of available class names.
474 Args:
475 operation: DisasterRecoveryClient method name e.g.
476 "create_dr_protection_group", "update_dr_protection_group",
477 "associate_dr_protection_group", "create_dr_plan_execution".
478 parameters: Keyword arguments for the SDK method.
479 profile: OCI config profile for the target region. Defaults to FSDR_REGION1.
481 Returns:
482 {
483 "data": ...,
484 "status": 200,
485 "headers": {...},
486 "work_request_id": "..." # present when the response carries one
487 }
488 """
489 if operation not in ALLOWED_FSDR_OPERATIONS:
490 raise ValueError(
491 f"Unsupported operation '{operation}'. "
492 f"Allowed: {', '.join(sorted(ALLOWED_FSDR_OPERATIONS))}"
493 )
495 client = get_dr_client(profile)
496 func = getattr(client, operation)
498 converted: Dict[str, Any] = {}
499 for key, value in parameters.items():
500 resolved = _resolve_model(value)
501 if key.endswith("_details") and isinstance(resolved, dict):
502 parts = key.split("_")[:-1]
503 class_name = "".join(p.capitalize() for p in parts) + "Details"
504 model_cls = getattr(oci.disaster_recovery.models, class_name, None)
505 if model_cls is not None: 505 ↛ 507line 505 didn't jump to line 507 because the condition on line 505 was always true
506 resolved = model_cls(**resolved)
507 converted[key] = resolved
509 resp = func(**converted)
510 result = _response_to_result(resp)
511 work_request_id = resp.headers.get("opc-work-request-id") if resp.headers else None
512 return WriteResult(
513 data=result.data,
514 status=result.status,
515 headers=result.headers,
516 work_request_id=work_request_id,
517 )
519# ---------------------------------------------------------------------------
520# Entry point
521# ---------------------------------------------------------------------------
524def main() -> None:
525 """Entry point for uvx / installed scripts.
527 MCP stdio uses stdout for JSON-RPC messages. Application logs are
528 configured above to use stderr so they do not corrupt the protocol stream.
529 """
530 mcp.run(show_banner=False)
533if __name__ == "__main__": 533 ↛ 534line 533 didn't jump to line 534 because the condition on line 533 was never true
534 main()