Set the correct API base URL (061b01a6) · Commits · OOP / code / ai2

mcp_module/tools/edge_application.py

+18 −15

Changes for mcp_module/tools/edge_application.py: 18 added lines, 15 removed lines.
Original line number	Diff line number	Diff line
		@@ -32,13 +32,6 @@ env_path = root_dir / ".env"
		load_dotenv(dotenv_path=env_path)


		def get_session_service_url() -> str:
		session_service_url = os.getenv("OEG_SERVICE_URL")
		if not session_service_url:
		raise OSError("OEG_SERVICE_URL environment variable is not set")
		return session_service_url


		def _derive_app_name(image_path: str) -> str:
		parsed = urlparse(image_path)
		image_ref = parsed.path if parsed.scheme and parsed.netloc else image_path
		@@ -50,13 +43,23 @@ def _derive_app_name(image_path: str) -> str:
		unique_id = uuid4().hex[:8]
		return f"{base_name}-{unique_id}"

		def get_edge_service_url() -> str:
		root_url = os.getenv("OEG_SERVICE_URL")
		if not root_url:
		raise EnvironmentError("OEG_SERVICE_URL environment variable is not set")

		return (
		f"{root_url.rstrip('/')}"
		"/edge-application-management/v0.1alpha1"
		)


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

		@@ -126,7 +129,7 @@ async def register_app_definition(inp: RegisterAppDefinitionInput) -> dict:
		if inp.memory is not None:
		required_resources["memory"] = inp.memory

		url = f"{get_session_service_url()}/apps"
		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))
		logger.debug("Calling endpoint: %s", url)
		@@ -155,7 +158,7 @@ async def delete_app_definition(inp: DeleteAppDefinitionInput) -> 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_session_service_url()}/apps/{inp.sessionId}"
		url = f"{get_edge_service_url()}/apps/{inp.sessionId}"
		logger.debug("Deleting app definition from: %s", url)

		try:
		@@ -183,7 +186,7 @@ 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}"
		url = f"{get_edge_service_url()}/apps/{inp.sessionId}"
		headers = {"accept": "application/json"}
		logger.debug("Fetching app definition from: %s", url)

		@@ -211,7 +214,7 @@ async def get_edge_cloud_zones() -> dict:
		"""
		Retrieve edge cloud zones
		"""
		url = f"{get_session_service_url()}/edge-cloud-zones"
		url = f"{get_edge_service_url()}/edge-cloud-zones"
		headers = {"accept": "application/json"}
		logger.debug("Fetching edge cloud zones from: %s", url)

		@@ -259,7 +262,7 @@ async def instantiate_app(inp: InstantiateAppInput) -> dict:
		app_zone["EdgeCloudZone"]["edgeCloudZoneId"] = edge_cloud_zone_id
		payload["appZones"].append(app_zone)

		url = f"{get_session_service_url()}/appinstances"
		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))
		logger.debug("Calling endpoint: %s", url)
		@@ -288,7 +291,7 @@ async def get_app_instances() -> dict:
		"""
		Retrieve instantiated or deployed applications information.
		"""
		url = f"{get_session_service_url()}/appinstances"
		url = f"{get_edge_service_url()}/appinstances"
		headers = {"accept": "application/json"}
		logger.debug("Fetching app instances from: %s", url)

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

		try:

ai_agent/requirements.txt

+1 −1

File changed.

Contains only whitespace changes.