docker_restart

Instructions

Implementation Reference

• src/opensips_mcp/tools/docker_tools.py:159-192 (handler)

  The docker_restart tool function: decorated with @mcp.tool(), @audited('docker_restart'), @require_permission('process.manage'). Executes 'docker compose restart <service>' as a subprocess with a 60-second timeout. Accepts a 'service' parameter (default 'opensips') and returns the exit code, stdout, stderr, and success flag.

  @mcp.tool()
  @audited("docker_restart")
  @require_permission("process.manage")
  async def docker_restart(
      ctx: Context,
      service: str = "opensips",
  ) -> dict[str, Any]:
      """Restart a Docker Compose service.
  
      Parameters
      ----------
      service:
          The service name to restart (default: ``opensips``).
      """
      try:
          proc = await asyncio.create_subprocess_exec(
              "docker", "compose", "restart", service,
              stdout=asyncio.subprocess.PIPE,
              stderr=asyncio.subprocess.PIPE,
          )
          stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60)
          return {
              "service": service,
              "returncode": proc.returncode,
              "output": stdout.decode().strip(),
              "errors": stderr.decode().strip() if proc.returncode != 0 else "",
              "success": proc.returncode == 0,
          }
      except FileNotFoundError:
          return {"error": "docker command not found -- is Docker installed?"}
      except asyncio.TimeoutError:
          return {"error": "docker compose restart timed out after 60 seconds"}
      except Exception as exc:
          return {"error": f"Failed to restart service: {exc}"}

• src/opensips_mcp/server.py:175-175 (registration)

  Registration of the docker_tools module on the MCP server. The import at line 175 in server.py triggers all @mcp.tool() decorators in docker_tools.py, including the docker_restart tool.

  from opensips_mcp.tools import docker_tools as _docker_tools  # noqa: E402, F401

• src/opensips_mcp/security/audit.py:45-63 (helper)

  The @audited() decorator used on docker_restart. It logs audit entries (success/error) for every call to the tool, recording the operation name ('docker_restart'), parameters, and the caller's role.

  def audited(operation: str):
      """Decorator that logs audit entries for tool calls."""
  
      def decorator(func):
          @wraps(func)
          async def wrapper(ctx: Context, *args, **kwargs):
              app = ctx.request_context.lifespan_context
              role = getattr(app.settings, "role", "unknown")
              try:
                  result = await func(ctx, *args, **kwargs)
                  audit_log(operation, kwargs, "success", role)
                  return result
              except Exception as e:
                  audit_log(operation, kwargs, f"error: {e}", role)
                  raise
  
          return wrapper
  
      return decorator

• src/opensips_mcp/security/rbac.py:93-130 (helper)

  The @require_permission() decorator used on docker_restart. It enforces RBAC by checking the caller's role has the 'process.manage' permission before allowing the tool to execute.

  def require_permission(permission: str) -> Callable[..., Any]:
      """Decorator that enforces RBAC on an MCP tool function.
  
      The decorated function **must** accept ``ctx: Context`` as its first
      positional argument.  The current role is read from
      ``ctx.request_context.lifespan_context.settings.role``.
  
      Raises ``McpError`` when the active role lacks the requested permission.
      """
  
      def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
          @functools.wraps(func)
          async def wrapper(ctx: Context, *args: Any, **kwargs: Any) -> Any:
              app_context = ctx.request_context.lifespan_context
              role_value: str = app_context.settings.role.lower()
  
              try:
                  role = Role(role_value)
              except ValueError:
                  raise McpError(
                      ErrorData(
                          code=-1,
                          message=f"Unknown role '{role_value}'. Valid roles: {[r.value for r in Role]}",
                      )
                  ) from None
  
              allowed = PERMISSIONS.get(role, set())
              if permission not in allowed:
                  raise McpError(
                      ErrorData(
                          code=-1,
                          message=f"Permission denied: role '{role.value}' lacks '{permission}' permission",
                      )
                  )
  
              return await func(ctx, *args, **kwargs)
  
          return wrapper