feat: added tools for edge zone retrieval and app instantiation,#9 (273378c0) · Commits · OOP / code / ai2

mcp_module/app.py

+4 −0

Original line number	Diff line number	Diff line
		@@ -10,8 +10,10 @@ from fastapi.responses import PlainTextResponse
		from dotenv import load_dotenv
		from mcp_module.tools.edge_application import (
		delete_app_definition,
		get_edge_cloud_zones,
		get_app_definitions,
		get_app_definition,
		instantiate_app,
		register_app_definition,
		)
		from mcp_module.tools.qod import (
		@@ -39,6 +41,8 @@ SAFE_TOOLS: list[Callable[..., Any]] = [
		register_app_definition,
		delete_app_definition,
		get_app_definition,
		get_edge_cloud_zones,
		instantiate_app,
		]

		for tool in SAFE_TOOLS:

mcp_module/json_templates/app_instantiate_request.json

0 → 100644

+14 −0

Original line number	Diff line number	Diff line
		{
		"appId": "",
		"appZones": [
		{
		"EdgeCloudZone": {
		"edgeCloudZoneId": "",
		"edgeCloudZoneName": "string",
		"edgeCloudZoneStatus": "unknown",
		"edgeCloudProvider": "string",
		"edgeCloudRegion": "string"
		}
		}
		]
		}

mcp_module/models/edge_application_models.py

+19 −0

Original line number	Diff line number	Diff line
		@@ -27,3 +27,22 @@ class DeleteAppDefinitionInput(BaseModel):

		class GetAppDefinitionInput(BaseModel):
		sessionId: str = Field(..., min_length=1)


		class InstantiateAppInput(BaseModel):
		appId: str = Field(..., min_length=1)
		edgeCloudZoneIds: list[str] = Field(..., min_length=1)

		@field_validator("edgeCloudZoneIds", mode="before")
		@classmethod
		def normalize_edge_cloud_zone_ids(cls, value):
		if isinstance(value, str):
		return [value]
		return value

		@field_validator("edgeCloudZoneIds")
		@classmethod
		def validate_edge_cloud_zone_ids(cls, value: list[str]) -> list[str]:
		if any(not zone_id.strip() for zone_id in value):
		raise ValueError("edgeCloudZoneIds must not contain empty values")
		return value

mcp_module/tools/edge_application.py

+85 −3

Original line number	Diff line number	Diff line
		import logging
		import json
		from copy import deepcopy
		import httpx
		from dotenv import load_dotenv
		from pathlib import Path
		@@ -9,6 +10,7 @@ from urllib.parse import urlparse
		from mcp_module.models.edge_application_models import (
		DeleteAppDefinitionInput,
		GetAppDefinitionInput,
		InstantiateAppInput,
		RegisterAppDefinitionInput,
		)

		@@ -34,9 +36,11 @@ def get_session_service_url() -> str:


		def _derive_app_name(image_path: str) -> str:
		parsed_path = urlparse(image_path).path
		last_segment = parsed_path.rstrip("/").split("/")[-1]
		name_without_tag = last_segment.split(":", 1)[0]
		parsed = urlparse(image_path)
		image_ref = parsed.path if parsed.scheme and parsed.netloc else image_path
		last_segment = image_ref.rstrip("/").split("/")[-1]
		name_without_digest = last_segment.split("@", 1)[0]
		name_without_tag = name_without_digest.split(":", 1)[0]
		if not name_without_tag:
		raise ValueError("imagePath must include an app name in the final URL segment")
		return name_without_tag
		@@ -186,3 +190,81 @@ async def get_app_definition(inp: GetAppDefinitionInput) -> dict:
		if isinstance(data, dict):
		return data
		return {"response": data}


		async def get_edge_cloud_zones() -> dict:
		"""
		Retrieve edge cloud zones with unknown status.
		"""
		url = f"{get_session_service_url()}/edge-cloud-zones"
		headers = {"accept": "application/json"}
		params = {"status": "unknown"}
		logger.debug("Fetching edge cloud zones from: %s", url)

		try:
		async with httpx.AsyncClient(trust_env=True) as client:
		response = await client.get(
		url,
		headers=headers,
		params=params,
		follow_redirects=True,
		timeout=15,
		)
		response.raise_for_status()
		except httpx.HTTPError as e:
		logger.error("Failed to fetch edge cloud zones: %s", e)
		raise

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

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


		async def instantiate_app(inp: InstantiateAppInput) -> dict:
		"""
		Instantiate or deploy an application to one or several Edge Cloud Zones.
		"""
		template_path = root_dir / "json_templates" / "app_instantiate_request.json"
		if not template_path.exists():
		raise FileNotFoundError(f"{template_path} not found")

		with open(template_path, "r", encoding="utf-8") as f:
		payload = json.load(f)

		zone_template = payload["appZones"][0]
		payload["appId"] = inp.appId
		payload["appZones"] = []

		for edge_cloud_zone_id in inp.edgeCloudZoneIds:
		app_zone = deepcopy(zone_template)
		app_zone["EdgeCloudZone"]["edgeCloudZoneId"] = edge_cloud_zone_id
		payload["appZones"].append(app_zone)

		url = f"{get_session_service_url()}/appinstances"
		headers = {"accept": "application/json", "content-type": "application/json"}
		logger.debug("Instantiate app payload: %s", json.dumps(payload, indent=2))
		logger.debug("Calling endpoint: %s", url)

		try:
		async with httpx.AsyncClient(trust_env=True) as client:
		response = await client.post(
		url, headers=headers, json=payload, follow_redirects=True, timeout=15
		)
		response.raise_for_status()
		except httpx.HTTPError as e:
		logger.error("Failed to instantiate app %s: %s", inp.appId, e)
		raise

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

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