carrierroute_replace_host

Instructions

Implementation Reference

• src/opensips_mcp/tools/carrierroute_tools.py:77-102 (handler)

  Async tool function that replaces a host in a carrier route at runtime. Calls MI command 'cr_replace_host' with carrier, domain, prefix, host, and new_host parameters. Decorated with @mcp.tool(), @audited('carrierroute_replace_host'), and @require_permission('mi.write').

  @mcp.tool()
  @audited("carrierroute_replace_host")
  @require_permission("mi.write")
  async def carrierroute_replace_host(
      ctx: Context,
      carrier: str,
      domain: str,
      prefix: str,
      host: str,
      new_host: str,
  ) -> dict[str, Any]:
      """Replace a host in carrier route (e.g. IP change).
  
      Does not require a reload; applies at runtime.
      """
      app = ctx.request_context.lifespan_context
      return await app.mi_client.execute(
          "cr_replace_host",
          {
              "carrier": carrier,
              "domain": domain,
              "prefix": prefix,
              "host": host,
              "new_host": new_host,
          },
      )

• src/opensips_mcp/tools/carrierroute_tools.py:1-18 (registration)

  Module imports: registers tools via @mcp.tool() decorator pattern on each async function. The 'mcp' instance is imported from opensips_mcp.server, which is a FastMCP instance (line 138 of server.py). The module is imported in server.py line 166 to trigger registration.

  """MCP tools for the ``carrierroute`` module (multi-carrier routing).
  
  Carrier routing provides prefix-based least-cost routing across multiple
  carriers with host activation/deactivation (maintenance windows, failover).
  """
  
  from __future__ import annotations
  
  import logging
  from typing import Any
  
  from mcp.server.fastmcp import Context
  
  from opensips_mcp.security.audit import audited
  from opensips_mcp.security.rbac import require_permission
  from opensips_mcp.server import mcp
  
  logger = logging.getLogger(__name__)

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

  FastMCP server instance creation and module import that triggers tool registration. The import of carrierroute_tools (line 166) causes all @mcp.tool() decorated functions to be registered.

  mcp = FastMCP("opensips-mcp-server", lifespan=app_lifespan)
  
  # Register tool modules (imported here to avoid circular imports).
  # Register prompt modules.
  from opensips_mcp.prompts import config_prompts as _cfg_prompts  # noqa: E402, F401
  
  # Register ecosystem prompt modules.
  from opensips_mcp.prompts import ecosystem_prompts as _eco_prompts  # noqa: E402, F401
  from opensips_mcp.prompts import educational_prompts as _edu_prompts  # noqa: E402, F401
  from opensips_mcp.prompts import migration_prompts as _mig_prompts  # noqa: E402, F401
  from opensips_mcp.prompts import security_prompts as _sec_prompts  # noqa: E402, F401
  from opensips_mcp.prompts import sip_prompts as _sip_prompts  # noqa: E402, F401
  from opensips_mcp.prompts import troubleshoot_prompts as _ts_prompts  # noqa: E402, F401
  from opensips_mcp.resources import config_resources as _cfg_res  # noqa: E402, F401
  from opensips_mcp.resources import db_resources as _db_res  # noqa: E402, F401
  from opensips_mcp.resources import docs_resources as _docs_res  # noqa: E402, F401
  
  # Register ecosystem resource modules.
  from opensips_mcp.resources import ecosystem_resources as _eco_res  # noqa: E402, F401
  from opensips_mcp.resources import scenario_resources as _scenario_res  # noqa: E402, F401
  
  # Register resource modules.
  from opensips_mcp.resources import system_resources as _sys_res  # noqa: E402, F401
  from opensips_mcp.tools import acc_tools as _acc_tools  # noqa: E402, F401
  from opensips_mcp.tools import avpops_tools as _avpops_tools  # noqa: E402, F401
  from opensips_mcp.tools import b2b_tools as _b2b_tools  # noqa: E402, F401
  from opensips_mcp.tools import benchmark_tools as _benchmark_tools  # noqa: E402, F401
  from opensips_mcp.tools import call_center_tools as _cc_tools  # noqa: E402, F401
  from opensips_mcp.tools import carrierroute_tools as _carrierroute_tools  # noqa: E402, F401

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

  The @audited decorator used on carrierroute_replace_host. Wraps the function to log audit entries on success/failure with operation name, params (masked), result status, and 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