feat: additional mcp tools for edge app API endpoints,#9 (d3155179) · Commits · OOP / code / ai2

mcp_module/app.py

+4 −0

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

		for tool in SAFE_TOOLS:

mcp_module/models/edge_application_models.py

+4 −0

Original line number	Diff line number	Diff line
		@@ -46,3 +46,7 @@ class InstantiateAppInput(BaseModel):
		if any(not zone_id.strip() for zone_id in value):
		raise ValueError("edgeCloudZoneIds must not contain empty values")
		return value


		class DeleteAppInstanceInput(BaseModel):
		appInstanceId: str = Field(..., min_length=1)

mcp_module/tools/edge_application.py

+63 −6

Original line number	Diff line number	Diff line
		import logging
		import json
		from copy import deepcopy
		from uuid import uuid4
		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,
		DeleteAppInstanceInput,
		GetAppDefinitionInput,
		InstantiateAppInput,
		RegisterAppDefinitionInput,
		@@ -40,10 +42,11 @@ def _derive_app_name(image_path: str) -> str:
		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:
		base_name = name_without_digest.split(":", 1)[0]
		if not base_name:
		raise ValueError("imagePath must include an app name in the final URL segment")
		return name_without_tag
		unique_id = uuid4().hex[:8]
		return f"{base_name}-{unique_id}"


		async def get_app_definitions() -> dict:
		@@ -194,11 +197,10 @@ async def get_app_definition(inp: GetAppDefinitionInput) -> dict:

		async def get_edge_cloud_zones() -> dict:
		"""
		Retrieve edge cloud zones with unknown status.
		Retrieve edge cloud zones
		"""
		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:
		@@ -206,7 +208,6 @@ async def get_edge_cloud_zones() -> dict:
		response = await client.get(
		url,
		headers=headers,
		params=params,
		follow_redirects=True,
		timeout=15,
		)
		@@ -268,3 +269,59 @@ async def instantiate_app(inp: InstantiateAppInput) -> dict:
		if isinstance(data, dict):
		return data
		return {"response": data}


		async def get_app_instances() -> dict:
		"""
		Retrieve instantiated or deployed applications information.
		"""
		url = f"{get_session_service_url()}/appinstances"
		headers = {"accept": "application/json"}
		logger.debug("Fetching app instances from: %s", url)

		try:
		async with httpx.AsyncClient(trust_env=True) as client:
		response = await client.get(
		url, headers=headers, follow_redirects=True, timeout=15
		)
		response.raise_for_status()
		except httpx.HTTPError as e:
		logger.error("Failed to fetch app instances: %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 {"appInstances": data}


		async def delete_app_instance(inp: DeleteAppInstanceInput) -> dict:
		"""
		Delete an instantiated or deployed application by app instance id.
		"""
		url = f"{get_session_service_url()}/appinstances/{inp.appInstanceId}"
		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.raise_for_status()
		except httpx.HTTPError as e:
		logger.error("Failed to delete app instance %s: %s", inp.appInstanceId, e)
		raise

		if not response.content:
		return {"status_code": response.status_code}

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

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