feat: add tools for edge apps definitions, #9 (7505ed15) · Commits · OOP / code / ai2

mcp_module/app.py

+9 −1

Original line number	Diff line number	Diff line
		@@ -8,7 +8,12 @@ from fastmcp import FastMCP
		from fastapi import FastAPI
		from fastapi.responses import PlainTextResponse
		from dotenv import load_dotenv
		from mcp_module.tools.edge_application import get_app_definitions
		from mcp_module.tools.edge_application import (
		delete_app_definition,
		get_app_definitions,
		get_app_definition,
		register_app_definition,
		)
		from mcp_module.tools.qod import (
		create_qod_session_oai,
		get_qod_session_oai,
		@@ -31,6 +36,9 @@ SAFE_TOOLS: list[Callable[..., Any]] = [
		get_qod_session_oai,
		delete_qod_session_oai,
		get_app_definitions,
		register_app_definition,
		delete_app_definition,
		get_app_definition,
		]

		for tool in SAFE_TOOLS:

mcp_module/json_templates/app_register_app_definition_request.json

0 → 100644

+22 −0

Original line number	Diff line number	Diff line
		{
		"name": "",
		"version": "1.0.0",
		"packageType": "",
		"appRepo": {
		"type": "PRIVATEREPO",
		"imagePath": ""
		},
		"componentSpec": [
		{
		"componentName": "",
		"networkInterfaces": [
		{
		"interfaceId": "eth0",
		"protocol": "TCP",
		"port": 0,
		"visibilityType": "VISIBILITY_EXTERNAL"
		}
		]
		}
		]
		}

mcp_module/models/edge_application_models.py

0 → 100644

+29 −0

Original line number	Diff line number	Diff line
		from typing import Literal

		from pydantic import BaseModel, Field, field_validator


		class RegisterAppDefinitionInput(BaseModel):
		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)

		@field_validator("packageType", mode="before")
		@classmethod
		def normalize_package_type(cls, value: str) -> str:
		normalized = str(value).strip().upper().replace("-", "_").replace(" ", "_")
		if normalized in {"HELM", "HELMCHART", "HELM_CHART"}:
		return "HELM"
		if normalized == "CONTAINER":
		return "CONTAINER"
		return normalized


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


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

mcp_module/tools/edge_application.py

+121 −0

Original line number	Diff line number	Diff line
		import logging
		import json
		import httpx
		from dotenv import load_dotenv
		from pathlib import Path
		import os
		from urllib.parse import urlparse

		from mcp_module.models.edge_application_models import (
		DeleteAppDefinitionInput,
		GetAppDefinitionInput,
		RegisterAppDefinitionInput,
		)

		# Configure logging
		logging.basicConfig(
		@@ -25,6 +33,15 @@ def get_session_service_url() -> str:
		return session_service_url


		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]
		if not name_without_tag:
		raise ValueError("imagePath must include an app name in the final URL segment")
		return name_without_tag


		async def get_app_definitions() -> dict:
		"""
		Get the list of all existing Application definitions from the Edge Cloud Provider
		@@ -65,3 +82,107 @@ async def get_app_definitions() -> dict:
		)
		return {}
		return data


		async def register_app_definition(inp: RegisterAppDefinitionInput) -> dict:
		"""
		Register an application definition from an image/chart URL and exposed port.

		Inputs:
		- packageType: HELM_CHART or CONTAINER
		- imagePath: image or Helm chart URL
		- port: exposed TCP port
		"""
		template_path = root_dir / "json_templates" / "app_register_app_definition_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)

		app_name = _derive_app_name(inp.imagePath)
		payload["name"] = app_name
		payload["packageType"] = inp.packageType
		payload["appRepo"]["imagePath"] = inp.imagePath
		payload["componentSpec"][0]["componentName"] = app_name
		payload["componentSpec"][0]["networkInterfaces"][0]["port"] = inp.port

		url = f"{get_session_service_url()}/apps"
		headers = {"accept": "application/json", "content-type": "application/json"}
		logger.debug("Register app definition 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 register app definition: %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 {"response": data}


		async def delete_app_definition(inp: DeleteAppDefinitionInput) -> dict:
		"""
		Delete all the information and content related to an Application.
		"""
		url = f"{get_session_service_url()}/apps/{inp.sessionId}"
		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.raise_for_status()
		except httpx.HTTPError as e:
		logger.error("Failed to delete app definition %s: %s", inp.sessionId, 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}


		async def get_app_definition(inp: GetAppDefinitionInput) -> dict:
		"""
		Retrieve the Application definition for a specific session id.
		"""
		url = f"{get_session_service_url()}/apps/{inp.sessionId}"
		headers = {"accept": "application/json"}
		logger.debug("Fetching app definition 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 definition %s: %s", inp.sessionId, 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}