drouting_delete_gateway

Instructions

Implementation Reference

• src/opensips_mcp/tools/drouting_tools.py:371-390 (handler)

  The actual tool handler for drouting_delete_gateway. It is decorated with @mcp.tool(), @audited('drouting_delete_gateway'), and @require_permission('db.write'). It accepts a gwid (int) and optional reload (bool), calls crud.delete_gateway to delete from the DB, and optionally triggers dr_reload via MI.

  @mcp.tool()
  @audited("drouting_delete_gateway")
  @require_permission("db.write")
  async def drouting_delete_gateway(
      ctx: Context,
      gwid: int,
      reload: bool = True,
  ) -> dict[str, Any]:
      """Delete a dynamic routing gateway by gwid."""
      from opensips_mcp.db.crud import drouting as crud
  
      app = ctx.request_context.lifespan_context
      async with app.db_session_factory() as session:
          deleted = await crud.delete_gateway(session, gwid)
      if not deleted:
          return {"error": "Gateway not found", "deleted": False}
      result: dict[str, Any] = {"gwid": gwid, "deleted": True}
      if reload:
          result.update(await _dr_reload(app))
      return result

• src/opensips_mcp/db/crud/drouting.py:156-160 (helper)

  The CRUD helper function delete_gateway that performs the actual deletion from the dr_gateways table. Uses SQLAlchemy delete statement on DRGateway model filtered by gwid, commits, and returns whether any row was deleted.

  async def delete_gateway(session: AsyncSession, gwid: int) -> bool:
      stmt = delete(DRGateway).where(DRGateway.gwid == gwid)
      result = await session.execute(stmt)
      await session.commit()
      return result.rowcount > 0

• src/opensips_mcp/db/models/drouting.py:27-39 (schema)

  The SQLAlchemy ORM model DRGateway, mapped to the 'dr_gateways' table. Defines columns: gwid (PK), type, address, strip, pri_prefix, attrs, probe_mode, state, socket, description.

  class DRGateway(Base):
      __tablename__ = "dr_gateways"
  
      gwid: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
      type: Mapped[int] = mapped_column(Integer, default=0)
      address: Mapped[str] = mapped_column(String(128), nullable=False)
      strip: Mapped[int] = mapped_column(Integer, default=0)
      pri_prefix: Mapped[str] = mapped_column(String(64), default="")
      attrs: Mapped[str] = mapped_column(String(255), default="")
      probe_mode: Mapped[int] = mapped_column(Integer, default=0)
      state: Mapped[int] = mapped_column(Integer, default=0)
      socket: Mapped[str] = mapped_column(String(128), default="")
      description: Mapped[str] = mapped_column(String(128), default="")

• src/opensips_mcp/tools/drouting_tools.py:17-24 (helper)

  The _dr_reload helper function that attempts to reload dynamic routing via MI command 'dr_reload'. Used optionally by the handler when reload=True.

  async def _dr_reload(app: Any) -> dict[str, Any]:
      """Attempt MI dr_reload and return status dict."""
      try:
          await app.mi_client.execute("dr_reload")
          return {"reloaded": True}
      except Exception as exc:
          logger.warning("dr_reload failed: %s", exc)
          return {"reloaded": False, "reload_error": str(exc)}