Skip to content

AI Ops API Package

ai_ops.api

REST API module for ai_ops app.

serializers

API serializers for ai_ops.

LLMMiddlewareSerializer

Bases: NautobotModelSerializer, TaggedModelSerializerMixin

LLMMiddleware Serializer.

Source code in ai_ops/api/serializers.py
class LLMMiddlewareSerializer(NautobotModelSerializer, TaggedModelSerializerMixin):  # pylint: disable=too-many-ancestors
    """LLMMiddleware Serializer."""

    class Meta:
        """Meta attributes."""

        model = models.LLMMiddleware
        fields = "__all__"
Meta

Meta attributes.

Source code in ai_ops/api/serializers.py
class Meta:
    """Meta attributes."""

    model = models.LLMMiddleware
    fields = "__all__"

LLMModelSerializer

Bases: NautobotModelSerializer, TaggedModelSerializerMixin

LLMModel Serializer.

Source code in ai_ops/api/serializers.py
class LLMModelSerializer(NautobotModelSerializer, TaggedModelSerializerMixin):  # pylint: disable=too-many-ancestors
    """LLMModel Serializer."""

    class Meta:
        """Meta attributes."""

        model = models.LLMModel
        fields = "__all__"
Meta

Meta attributes.

Source code in ai_ops/api/serializers.py
class Meta:
    """Meta attributes."""

    model = models.LLMModel
    fields = "__all__"

LLMProviderSerializer

Bases: NautobotModelSerializer, TaggedModelSerializerMixin

LLMProvider Serializer.

Source code in ai_ops/api/serializers.py
class LLMProviderSerializer(NautobotModelSerializer, TaggedModelSerializerMixin):  # pylint: disable=too-many-ancestors
    """LLMProvider Serializer."""

    class Meta:
        """Meta attributes."""

        model = models.LLMProvider
        fields = "__all__"
Meta

Meta attributes.

Source code in ai_ops/api/serializers.py
class Meta:
    """Meta attributes."""

    model = models.LLMProvider
    fields = "__all__"

MCPServerSerializer

Bases: NautobotModelSerializer, TaggedModelSerializerMixin

MCPServer Serializer.

Source code in ai_ops/api/serializers.py
class MCPServerSerializer(NautobotModelSerializer, TaggedModelSerializerMixin):  # pylint: disable=too-many-ancestors
    """MCPServer Serializer."""

    class Meta:
        """Meta attributes."""

        model = models.MCPServer
        fields = "__all__"
Meta

Meta attributes.

Source code in ai_ops/api/serializers.py
class Meta:
    """Meta attributes."""

    model = models.MCPServer
    fields = "__all__"

MiddlewareTypeSerializer

Bases: NautobotModelSerializer, TaggedModelSerializerMixin

MiddlewareType Serializer.

Source code in ai_ops/api/serializers.py
class MiddlewareTypeSerializer(NautobotModelSerializer, TaggedModelSerializerMixin):  # pylint: disable=too-many-ancestors
    """MiddlewareType Serializer."""

    class Meta:
        """Meta attributes."""

        model = models.MiddlewareType
        fields = "__all__"
Meta

Meta attributes.

Source code in ai_ops/api/serializers.py
class Meta:
    """Meta attributes."""

    model = models.MiddlewareType
    fields = "__all__"

SystemPromptSerializer

Bases: NautobotModelSerializer, TaggedModelSerializerMixin

SystemPrompt Serializer.

Source code in ai_ops/api/serializers.py
class SystemPromptSerializer(NautobotModelSerializer, TaggedModelSerializerMixin):  # pylint: disable=too-many-ancestors
    """SystemPrompt Serializer."""

    class Meta:
        """Meta attributes."""

        model = models.SystemPrompt
        fields = "__all__"
        read_only_fields = ["version"]  # Version auto-increments when prompt_text is updated
Meta

Meta attributes.

Source code in ai_ops/api/serializers.py
class Meta:
    """Meta attributes."""

    model = models.SystemPrompt
    fields = "__all__"
    read_only_fields = ["version"]  # Version auto-increments when prompt_text is updated

urls

Django API urlpatterns declaration for ai_ops app.

views

API views for ai_ops.

LLMMiddlewareViewSet

Bases: NautobotModelViewSet

LLMMiddleware viewset.

Source code in ai_ops/api/views.py
class LLMMiddlewareViewSet(NautobotModelViewSet):  # pylint: disable=too-many-ancestors
    """LLMMiddleware viewset."""

    queryset = models.LLMMiddleware.objects.all()
    serializer_class = serializers.LLMMiddlewareSerializer
    filterset_class = filters.LLMMiddlewareFilterSet

LLMModelViewSet

Bases: NautobotModelViewSet

LLMModel viewset.

Source code in ai_ops/api/views.py
class LLMModelViewSet(NautobotModelViewSet):  # pylint: disable=too-many-ancestors
    """LLMModel viewset."""

    queryset = models.LLMModel.objects.all()
    serializer_class = serializers.LLMModelSerializer
    filterset_class = filters.LLMModelFilterSet

LLMProviderViewSet

Bases: NautobotModelViewSet

LLMProvider viewset.

Source code in ai_ops/api/views.py
class LLMProviderViewSet(NautobotModelViewSet):  # pylint: disable=too-many-ancestors
    """LLMProvider viewset."""

    queryset = models.LLMProvider.objects.all()
    serializer_class = serializers.LLMProviderSerializer
    filterset_class = filters.LLMProviderFilterSet

MCPServerViewSet

Bases: NautobotModelViewSet

MCPServer viewset.

Source code in ai_ops/api/views.py
class MCPServerViewSet(NautobotModelViewSet):  # pylint: disable=too-many-ancestors
    """MCPServer viewset."""

    queryset = models.MCPServer.objects.all()
    serializer_class = serializers.MCPServerSerializer
    filterset_class = filters.MCPServerFilterSet

    @action(detail=True, methods=["post"], url_path="health-check")
    def health_check(self, request, pk=None):
        """Perform health check on MCP server."""
        mcp_server = self.get_object()

        try:
            # Build health check URL using base URL + health_check path
            # Note: health check is NOT at the MCP endpoint, it's at the base URL
            health_path = getattr(mcp_server, "health_check", "/health")
            health_url = f"{mcp_server.url.rstrip('/')}{health_path}"

            # Only disable SSL verification for internal MCP servers
            verify_ssl = mcp_server.mcp_type != "internal"

            # Perform health check with sync client
            with httpx.Client(verify=verify_ssl, timeout=5.0) as client:
                response = client.get(health_url)

                if response.status_code == 200:
                    return Response(
                        {
                            "success": True,
                            "message": f"MCP Server '{mcp_server.name}' is healthy",
                            "details": f"Successfully connected to {health_url}",
                            "url": health_url,
                        }
                    )
                else:
                    return Response(
                        {
                            "success": False,
                            "message": f"MCP Server '{mcp_server.name}' health check failed",
                            "details": f"HTTP {response.status_code} from {health_url}",
                            "url": health_url,
                        }
                    )
        except httpx.TimeoutException:
            health_path = getattr(mcp_server, "health_check", "/health")
            health_url = f"{mcp_server.url.rstrip('/')}{health_path}"
            return Response(
                {
                    "success": False,
                    "message": f"MCP Server '{mcp_server.name}' health check timed out",
                    "details": f"No response after 5 seconds from {health_url}",
                    "url": health_url,
                }
            )
        except Exception as e:
            health_path = getattr(mcp_server, "health_check", "/health")
            health_url = f"{mcp_server.url.rstrip('/')}{health_path}"

            # Only hide exception details in NONPROD and PROD environments for security
            env = get_environment()
            if env not in (NautobotEnvironment.NONPROD, NautobotEnvironment.PROD):
                error_details = str(e)
            else:
                error_details = "Connection error. Please check server configuration."

            return Response(
                {
                    "success": False,
                    "message": f"MCP Server '{mcp_server.name}' health check failed",
                    "details": error_details,
                    "url": health_url,
                }
            )
health_check(request, pk=None)

Perform health check on MCP server.

Source code in ai_ops/api/views.py
@action(detail=True, methods=["post"], url_path="health-check")
def health_check(self, request, pk=None):
    """Perform health check on MCP server."""
    mcp_server = self.get_object()

    try:
        # Build health check URL using base URL + health_check path
        # Note: health check is NOT at the MCP endpoint, it's at the base URL
        health_path = getattr(mcp_server, "health_check", "/health")
        health_url = f"{mcp_server.url.rstrip('/')}{health_path}"

        # Only disable SSL verification for internal MCP servers
        verify_ssl = mcp_server.mcp_type != "internal"

        # Perform health check with sync client
        with httpx.Client(verify=verify_ssl, timeout=5.0) as client:
            response = client.get(health_url)

            if response.status_code == 200:
                return Response(
                    {
                        "success": True,
                        "message": f"MCP Server '{mcp_server.name}' is healthy",
                        "details": f"Successfully connected to {health_url}",
                        "url": health_url,
                    }
                )
            else:
                return Response(
                    {
                        "success": False,
                        "message": f"MCP Server '{mcp_server.name}' health check failed",
                        "details": f"HTTP {response.status_code} from {health_url}",
                        "url": health_url,
                    }
                )
    except httpx.TimeoutException:
        health_path = getattr(mcp_server, "health_check", "/health")
        health_url = f"{mcp_server.url.rstrip('/')}{health_path}"
        return Response(
            {
                "success": False,
                "message": f"MCP Server '{mcp_server.name}' health check timed out",
                "details": f"No response after 5 seconds from {health_url}",
                "url": health_url,
            }
        )
    except Exception as e:
        health_path = getattr(mcp_server, "health_check", "/health")
        health_url = f"{mcp_server.url.rstrip('/')}{health_path}"

        # Only hide exception details in NONPROD and PROD environments for security
        env = get_environment()
        if env not in (NautobotEnvironment.NONPROD, NautobotEnvironment.PROD):
            error_details = str(e)
        else:
            error_details = "Connection error. Please check server configuration."

        return Response(
            {
                "success": False,
                "message": f"MCP Server '{mcp_server.name}' health check failed",
                "details": error_details,
                "url": health_url,
            }
        )

MiddlewareTypeViewSet

Bases: NautobotModelViewSet

MiddlewareType viewset.

Source code in ai_ops/api/views.py
class MiddlewareTypeViewSet(NautobotModelViewSet):  # pylint: disable=too-many-ancestors
    """MiddlewareType viewset."""

    queryset = models.MiddlewareType.objects.all()
    serializer_class = serializers.MiddlewareTypeSerializer
    filterset_class = filters.MiddlewareTypeFilterSet

    @action(detail=True, methods=["get"], url_path="default-config")
    def default_config(self, request: Request, pk: Optional[str] = None) -> Response:
        """Get the default configuration for a specific middleware type.

        Args:
            request: HTTP request object
            pk: Primary key of the middleware type

        Returns:
            Response: JSON response with default configuration for the middleware type
        """
        middleware_type = self.get_object()
        default_config = get_default_config_for_middleware(middleware_type.name)

        return Response(
            {
                "middleware_type": middleware_type.name,
                "default_config": default_config,
            }
        )
default_config(request, pk=None)

Get the default configuration for a specific middleware type.

Parameters:

Name Type Description Default
request Request

HTTP request object

required
pk Optional[str]

Primary key of the middleware type

None

Returns:

Name Type Description
Response Response

JSON response with default configuration for the middleware type

Source code in ai_ops/api/views.py
@action(detail=True, methods=["get"], url_path="default-config")
def default_config(self, request: Request, pk: Optional[str] = None) -> Response:
    """Get the default configuration for a specific middleware type.

    Args:
        request: HTTP request object
        pk: Primary key of the middleware type

    Returns:
        Response: JSON response with default configuration for the middleware type
    """
    middleware_type = self.get_object()
    default_config = get_default_config_for_middleware(middleware_type.name)

    return Response(
        {
            "middleware_type": middleware_type.name,
            "default_config": default_config,
        }
    )

SystemPromptViewSet

Bases: NautobotModelViewSet

SystemPrompt viewset.

Source code in ai_ops/api/views.py
class SystemPromptViewSet(NautobotModelViewSet):  # pylint: disable=too-many-ancestors
    """SystemPrompt viewset."""

    queryset = models.SystemPrompt.objects.all()
    serializer_class = serializers.SystemPromptSerializer
    filterset_class = filters.SystemPromptFilterSet