Commit f9ae4c8d authored by Natalia Grigoriadou's avatar Natalia Grigoriadou
Browse files

updated edge application tools

parent dfde17d0
Loading
Loading
Loading
Loading
+2 −0
Changes for mcp_module/app.py: 2 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -12,6 +12,7 @@ from fastmcp import FastMCP
from mcp_module.tools.edge_application import (
    create_app_deployment,
    delete_app_definition,
    delete_app_deployment,
    delete_app_instance,
    get_app_definition,
    get_app_definitions,
@@ -52,6 +53,7 @@ SAFE_TOOLS: list[Callable[..., Any]] = [
    get_app_instances,
    delete_app_instance,
    create_app_deployment,
    delete_app_deployment
]

for tool in SAFE_TOOLS:
+2 −17
Changes for mcp_module/json_templates/app_register_app_definition_request.json: 2 added lines, 17 removed lines.
Original line number Diff line number Diff line
@@ -2,23 +2,8 @@
  "name": "",
  "version": "1.0.0",
  "packageType": "",
  "appRepo": {
    "type": "PRIVATEREPO",
    "imagePath": ""
  },
  "appRepo": {},
  "appProvider": "",
  "requiredResources": {}, 
  "componentSpec": [
    {
      "componentName": "",
      "networkInterfaces": [
        {
          "interfaceId": "eth0",
          "protocol": "TCP",
          "port": 0,
          "visibilityType": "VISIBILITY_EXTERNAL"
        }
      ]
    }
  ]
  "componentSpec": []
}
+69 −3
Changes for mcp_module/models/edge_application_models.py: 69 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -3,7 +3,53 @@ from typing import Literal
from pydantic import BaseModel, Field, field_validator


class AppRepoInput(BaseModel):
    type: Literal["PRIVATEREPO", "PUBLICREPO"]
    imagePath: str = Field(..., max_length=2048)
    userName: str | None = Field(default=None, max_length=64)
    credentials: str | None = Field(default=None, max_length=2048)
    authType: Literal["DOCKER", "HTTP_BASIC", "HTTP_BEARER", "NONE"] | None = None
    checksum: str | None = Field(default=None, max_length=128)
class NetworkInterfaceInput(BaseModel):
    interfaceId: str = Field(
        ...,
        min_length=4,
        max_length=32,
        pattern=r"^[A-Za-z0-9][A-Za-z0-9_]{2,30}[A-Za-z0-9]$",
    )
    protocol: Literal["TCP", "UDP", "ANY"]
    port: int = Field(..., ge=1, le=65535)
    visibilityType: Literal["VISIBILITY_EXTERNAL", "VISIBILITY_INTERNAL"]


class ComponentSpecInput(BaseModel):
    componentName: str = Field(..., max_length=64)
    networkInterfaces: list[NetworkInterfaceInput] = Field(
        ...,
        min_length=1,
        max_length=100,
    )

class OperatingSystemInput(BaseModel):
    architecture: Literal["x86_64", "x86"]
    family: Literal["RHEL", "UBUNTU", "COREOS", "WINDOWS", "OTHER"]
    version: Literal[
        "OS_VERSION_UBUNTU_2204_LTS",
        "OS_VERSION_RHEL_8",
        "OS_MS_WINDOWS_2022",
        "OTHER",
    ]
    license: Literal[
        "OS_LICENSE_TYPE_FREE",
        "OS_LICENSE_TYPE_ON_DEMAND",
        "OTHER",
    ]
class RegisterAppDefinitionInput(BaseModel):
    appId: str | None = Field(
        default=None,
        pattern=r"^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$",
        description="Optional application UUID. OEG generates one when omitted.",
    )
    name: str = Field(
        ...,
        max_length=64,
@@ -14,13 +60,13 @@ class RegisterAppDefinitionInput(BaseModel):
        max_length=64,
        pattern=r"^[A-Za-z][A-Za-z0-9_]{7,63}$",
    )
    appRepo: AppRepoInput
    version: str = Field(default="1.0.0", max_length=64)
    repoType: Literal["PUBLICREPO", "PRIVATEREPO"]
    packageType: Literal["HELM", "CONTAINER"] = Field(
        ..., description="Allowed package types: HELM or CONTAINER"
    )
    imagePath: str = Field(..., min_length=1)
    port: int = Field(..., ge=1, le=65535)
    operatingSystem: OperatingSystemInput | None = None
    componentSpec: list[ComponentSpecInput] = Field(..., max_length=100)
    requiredResources: dict = Field(
        ...,
        description=(
@@ -60,6 +106,14 @@ class InstantiateAppInput(BaseModel):
    edgeCloudZoneId: str = Field(..., min_length=1)
    kubernetesClusterRef: str | None = None
    subscriptionRequest: dict | None = None
    idempotencyKey: str | None = Field(
        default=None,
        max_length=128,
        description=(
            "Optional key identifying this creation request. "
            "Reuse it when retrying the same operation."
        ),
    )


class DeleteAppInstanceInput(BaseModel):
@@ -79,3 +133,15 @@ class CreateAppDeploymentInput(BaseModel):
        max_length=100,
    )
    subscriptionRequest: dict | None = None
    idempotencyKey: str | None = Field(
        default=None,
        max_length=128,
        description=(
            "Optional key identifying this creation request. "
            "Reuse it when retrying the same operation."
        ),
    )


class DeleteAppDeploymentInput(BaseModel):
    appDeploymentId: str = Field(..., min_length=1)
 No newline at end of file
+126 −28
Changes for mcp_module/tools/edge_application.py: 126 added lines, 28 removed lines.
Original line number Diff line number Diff line
@@ -4,13 +4,14 @@ import logging
import os
from pathlib import Path
from typing import Literal

import re
import httpx
from dotenv import load_dotenv

from mcp_module.models.edge_application_models import (
    CreateAppDeploymentInput,
    DeleteAppDefinitionInput,
    DeleteAppDeploymentInput,
    DeleteAppInstanceInput,
    GetAppDefinitionInput,
    InstantiateAppInput,
@@ -31,6 +32,16 @@ env_path = root_dir / ".env"
load_dotenv(dotenv_path=env_path)


def get_edge_headers(x_correlator: str | None = None) -> dict[str, str]:
    headers = {"accept": "application/json"}

    if x_correlator is not None:
        if not re.fullmatch(r"[a-zA-Z0-9-_:;.\\/<>{}]{0,256}", x_correlator):
            raise ValueError("Invalid x-correlator value")
        headers["x-correlator"] = x_correlator

    return headers


def get_edge_service_url() -> str:
    root_url = os.getenv("OEG_SERVICE_URL")
@@ -40,13 +51,15 @@ def get_edge_service_url() -> str:
    return f"{root_url.rstrip('/')}/edge-application-management/v0.1alpha1"


async def get_app_definitions() -> dict:
async def get_app_definitions(
    xCorrelator: str | None = None,
) -> dict:
    """
    Get the list of all existing Application definitions from the Edge Cloud Provider
    that the user has permission to view.
    """
    url = f"{get_edge_service_url()}/apps"
    headers = {"accept": "application/json"}
    headers = get_edge_headers(xCorrelator)
    logger.debug("Fetching edge application definitions from: %s", url)

    try:
@@ -77,12 +90,15 @@ async def get_app_definitions() -> dict:
    return {"applications": data}


async def register_app_definition(inp: RegisterAppDefinitionInput) -> dict:
async def register_app_definition(
    inp: RegisterAppDefinitionInput,
    xCorrelator: str | None = None,
) -> dict:
    """
    Register an application manifest with the Edge Cloud Provider.

    Supply the application name, provider, version, package and repository
    details, exposed port, and Kubernetes resource requirements.
    details, components and network interfaces, and Kubernetes resources.

    Returns the appId assigned by OEG.
    """
@@ -95,20 +111,23 @@ async def register_app_definition(inp: RegisterAppDefinitionInput) -> dict:
    payload = json.loads(
        await asyncio.to_thread(template_path.read_text, encoding="utf-8")
    )

    if inp.appId is not None:
        payload["appId"] = inp.appId
    if inp.operatingSystem is not None:
        payload["operatingSystem"] = inp.operatingSystem.model_dump()
    payload["name"] = inp.name
    payload["appProvider"] = inp.appProvider
    payload["version"] = inp.version
    payload["packageType"] = inp.packageType
    payload["appRepo"]["type"] = inp.repoType
    payload["appRepo"]["imagePath"] = inp.imagePath
    payload["appRepo"] = inp.appRepo.model_dump(exclude_none=True)
    payload["requiredResources"] = inp.requiredResources
    payload["componentSpec"][0]["componentName"] = inp.name
    payload["componentSpec"][0]["networkInterfaces"][0]["port"] = inp.port

    payload["componentSpec"] = [
        component.model_dump() for component in inp.componentSpec
    ]
    url = f"{get_edge_service_url()}/apps"
    headers = {"accept": "application/json", "content-type": "application/json"}
    logger.debug("Register app definition payload: %s", json.dumps(payload, indent=2))
    headers = get_edge_headers(xCorrelator)
    headers["content-type"] = "application/json"
    logger.debug("Registering application definition: %s", inp.name)
    logger.debug("Calling endpoint: %s", url)

    try:
@@ -131,16 +150,22 @@ async def register_app_definition(inp: RegisterAppDefinitionInput) -> dict:
    return {"response": data}


async def delete_app_definition(inp: DeleteAppDefinitionInput) -> dict:
async def delete_app_definition(
    inp: DeleteAppDefinitionInput,
    xCorrelator: str | None = None,
) -> dict:
    """
    Delete all the information and content related to an application definition id. If there are multiple ones, invoke the tool multiple times.
    """
    url = f"{get_edge_service_url()}/apps/{inp.appId}"
    headers = get_edge_headers(xCorrelator)
    logger.debug("Deleting app definition from: %s", url)

    try:
        async with httpx.AsyncClient(trust_env=True) as client:
            response = await client.delete(url, follow_redirects=True, timeout=15)
            response = await client.delete(
                url, headers=headers, follow_redirects=True, timeout=15
            )
            response.raise_for_status()
    except httpx.HTTPError as e:
        logger.error("Failed to delete app definition %s: %s", inp.appId, e)
@@ -159,12 +184,15 @@ async def delete_app_definition(inp: DeleteAppDefinitionInput) -> dict:
    return {"response": data}


async def get_app_definition(inp: GetAppDefinitionInput) -> dict:
async def get_app_definition(
    inp: GetAppDefinitionInput,
    xCorrelator: str | None = None,
) -> dict:
    """
    Retrieve the Application definition for a specific app id.
    """
    url = f"{get_edge_service_url()}/apps/{inp.appId}"
    headers = {"accept": "application/json"}
    headers = get_edge_headers(xCorrelator)
    logger.debug("Fetching app definition from: %s", url)

    try:
@@ -194,12 +222,13 @@ async def get_app_definition(inp: GetAppDefinitionInput) -> dict:
async def get_edge_cloud_zones(
    region: str | None = None,
    status: Literal["active", "inactive", "unknown"] | None = None,
    xCorrelator: str | None = None,
) -> dict:
    """
    Retrieve edge cloud zones, optionally filtered by region and status.
    """
    url = f"{get_edge_service_url()}/edge-cloud-zones"
    headers = {"accept": "application/json"}
    headers = get_edge_headers(xCorrelator)

    params = {}
    if region is not None:
@@ -233,9 +262,12 @@ async def get_edge_cloud_zones(
    return {"edgeCloudZones": data}


async def instantiate_app(inp: InstantiateAppInput) -> dict:
async def instantiate_app(
    inp: InstantiateAppInput,
    xCorrelator: str | None = None,
) -> dict:
    """
    Instantiate or deploy an application to one or several Edge Cloud Zones.
    Instantiate or deploy an application to a single edge cloud zone.
    """
    template_path = root_dir / "json_templates" / "app_instantiate_request.json"
    if not template_path.exists():
@@ -256,8 +288,11 @@ async def instantiate_app(inp: InstantiateAppInput) -> dict:
        payload["subscriptionRequest"] = inp.subscriptionRequest

    url = f"{get_edge_service_url()}/appinstances"
    headers = {"accept": "application/json", "content-type": "application/json"}
    logger.debug("Instantiate app payload: %s", json.dumps(payload, indent=2))
    headers = get_edge_headers(xCorrelator)
    headers["content-type"] = "application/json"
    if inp.idempotencyKey is not None:
        headers["Idempotency-Key"] = inp.idempotencyKey
    logger.debug("Requesting instantiation for application: %s", inp.appId)
    logger.debug("Calling endpoint: %s", url)

    try:
@@ -284,12 +319,13 @@ async def get_app_instances(
    appId: str | None = None,
    appInstanceId: str | None = None,
    region: str | None = None,
    xCorrelator: str | None = None,
) -> dict:
    """
    Retrieve instantiated or deployed applications information.
    """
    url = f"{get_edge_service_url()}/appinstances"
    headers = {"accept": "application/json"}
    headers = get_edge_headers(xCorrelator)
    params = {}
    if appId is not None:
        params["appId"] = appId
@@ -319,16 +355,22 @@ async def get_app_instances(
    return {"appInstances": data}


async def delete_app_instance(inp: DeleteAppInstanceInput) -> dict:
async def delete_app_instance(
    inp: DeleteAppInstanceInput,
    xCorrelator: str | None = None,
) -> dict:
    """
    Delete an instantiated or deployed application by app instance id. If there are multiple ones, invoke the tool multiple times.
    """
    url = f"{get_edge_service_url()}/appinstances/{inp.appInstanceId}"
    headers = get_edge_headers(xCorrelator)
    logger.debug("Deleting app instance from: %s", url)

    try:
        async with httpx.AsyncClient(trust_env=True) as client:
            response = await client.delete(url, follow_redirects=True, timeout=15)
            response = await client.delete(
                url, headers=headers, follow_redirects=True, timeout=15
            )
            response.raise_for_status()
    except httpx.HTTPError as e:
        logger.error("Failed to delete app instance %s: %s", inp.appInstanceId, e)
@@ -350,7 +392,10 @@ async def delete_app_instance(inp: DeleteAppInstanceInput) -> dict:
    return {"response": data}


async def create_app_deployment(inp: CreateAppDeploymentInput) -> dict:
async def create_app_deployment(
    inp: CreateAppDeploymentInput,
    xCorrelator: str | None = None,
) -> dict:
    """
    Request application deployment across one or more edge cloud zones.

@@ -374,7 +419,10 @@ async def create_app_deployment(inp: CreateAppDeploymentInput) -> dict:
        payload["subscriptionRequest"] = inp.subscriptionRequest

    url = f"{get_edge_service_url()}/deployments"
    headers = {"accept": "application/json", "content-type": "application/json"}
    headers = get_edge_headers(xCorrelator)
    headers["content-type"] = "application/json"
    if inp.idempotencyKey is not None:
        headers["Idempotency-Key"] = inp.idempotencyKey
    logger.debug("Creating application deployment at: %s", url)

    try:
@@ -409,16 +457,18 @@ async def create_app_deployment(inp: CreateAppDeploymentInput) -> dict:
        "message": "Application deployment accepted; completion is pending.",
    }


async def get_app_deployments(
    appId: str | None = None,
    appDeploymentId: str | None = None,
    xCorrelator: str | None = None,
) -> dict:
    """
    List application deployments visible to the authenticated caller.

    """
    url = f"{get_edge_service_url()}/deployments"
    headers = {"accept": "application/json"}
    headers = get_edge_headers(xCorrelator)

    params = {}
    if appId is not None:
@@ -452,3 +502,51 @@ async def get_app_deployments(
        raise TypeError("Expected a list of application deployments from OEG")

    return {"appDeployments": data}


async def delete_app_deployment(
    inp: DeleteAppDeploymentInput,
    xCorrelator: str | None = None,
) -> dict:
    """
    Request termination of an application deployment by appDeploymentId.

    HTTP 202 means termination was accepted and may still be pending.
    """
    url = f"{get_edge_service_url()}/deployments/{inp.appDeploymentId}"
    headers = get_edge_headers(xCorrelator)
    logger.debug("Deleting application deployment from: %s", url)

    try:
        async with httpx.AsyncClient(trust_env=True) as client:
            response = await client.delete(
                url,
                headers=headers,
                follow_redirects=True,
                timeout=15,
            )
            response.raise_for_status()
    except httpx.HTTPError as e:
        logger.error(
            "Failed to delete application deployment %s: %s",
            inp.appDeploymentId,
            e,
        )
        raise

    if not response.content:
        return {
            "status_code": response.status_code,
            "message": (
                "Application deployment termination accepted; completion is pending."
            ),
        }

    try:
        data = response.json()
    except ValueError:
        return {"status_code": response.status_code, "body": response.text}

    if isinstance(data, dict):
        return data
    return {"response": data}