added trafic influence apis (4e8e0f5c) · Commits · OOP / code / ai2

mcp_module/app.py

+11 −1

Changes for mcp_module/app.py: 11 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -27,6 +27,12 @@ from mcp_module.tools.qod import (
		delete_qod_session_oai,
		get_qod_session_oai,
		)
		from mcp_module.tools.traffic_influence import (
		create_traffic_influence,
		delete_traffic_influence,
		get_traffic_influence,
		get_traffic_influences,
		)

		logging.basicConfig(
		level=logging.DEBUG, format="%(asctime)s [%(levelname)s] %(message)s"
		@@ -53,7 +59,11 @@ SAFE_TOOLS: list[Callable[..., Any]] = [
		get_app_instances,
		delete_app_instance,
		create_app_deployment,
		delete_app_deployment
		delete_app_deployment,
		create_traffic_influence,
		get_traffic_influences,
		get_traffic_influence,
		delete_traffic_influence,
		]

		for tool in SAFE_TOOLS:

mcp_module/models/qod_models.py

+125 −44

Changes for mcp_module/models/qod_models.py: 125 added lines, 44 removed lines.
Original line number	Diff line number	Diff line
		import re
		from ipaddress import IPv6Address
		from typing import Any
		from datetime import datetime, timezone
		from ipaddress import IPv4Address as IPv4AddressValue
		from ipaddress import IPv6Address as IPv6AddressValue
		from typing import Any, Literal

		from pydantic import BaseModel, Field, field_validator, model_validator
		from pydantic import AwareDatetime, BaseModel, Field, field_validator, model_validator


		class IPv4Address(BaseModel):
		publicAddress: str
		publicPort: int
		privateAddress: str \| None = None
		publicPort: int \| None = Field(default=None, ge=0, le=65535)

		@field_validator("publicAddress", "privateAddress")
		@classmethod
		def validate_address(cls, value: str \| None) -> str \| None:
		if value is not None:
		IPv4AddressValue(value)
		return value

		E164_REGEX = re.compile(r"^\+[1-9]\d{1,14}$")
		@model_validator(mode="after")
		def require_private_address_or_port(self):
		if self.privateAddress is None and self.publicPort is None:
		raise ValueError("Provide privateAddress or publicPort")
		return self


		class DeviceInput(BaseModel):
		phoneNumber: str \| None = None
		phoneNumber: str \| None = Field(
		default=None,
		pattern=r"^\+[1-9][0-9]{4,14}$",
		)
		networkAccessIdentifier: str \| None = None
		ipv4Address: IPv4Address \| None = None
		ipv6Address: IPv6Address \| None = None
		duration: int \| None = Field(
		None, ge=60, le=86400, description="Duration in seconds (60-86400)"
		)
		ipv6Address: str \| None = None

		@field_validator("phoneNumber")
		@field_validator("ipv6Address")
		@classmethod
		def validate_phone_number(cls, v):
		if v is not None and not E164_REGEX.match(v):
		raise ValueError(
		"phoneNumber must conform to E.164 format (e.g., +1234567890)"
		def validate_ipv6(cls, value: str \| None) -> str \| None:
		if value is not None:
		IPv6AddressValue(value)
		return value

		@model_validator(mode="after")
		def require_device_identifier(self):
		if not any(
		(
		self.phoneNumber,
		self.networkAccessIdentifier,
		self.ipv4Address,
		self.ipv6Address,
		)
		return v
		):
		raise ValueError("Provide at least one device identifier")
		return self


		class ApplicationServerInput(BaseModel):
		ipv4Address: str \| None = None
		ipv6Address: str \| None = None

		@model_validator(mode="after")
		def require_server_address(self):
		if not self.ipv4Address and not self.ipv6Address:
		raise ValueError("Provide at least one application-server address")
		return self


		class PortRangeInput(BaseModel):
		from_: int = Field(alias="from", ge=0, le=65535)
		to: int = Field(..., ge=0, le=65535)


		class PortsSpecInput(BaseModel):
		ranges: list[PortRangeInput] \| None = Field(default=None, min_length=1)
		ports: list[int] \| None = Field(default=None, min_length=1)

		@model_validator(mode="after")
		def require_ranges_or_ports(self):
		if not self.ranges and not self.ports:
		raise ValueError("Provide port ranges or individual ports")
		return self


		class SinkCredentialInput(BaseModel):
		credentialType: Literal["PLAIN", "ACCESSTOKEN", "REFRESHTOKEN"]
		accessToken: str \| None = Field(default=None, repr=False)
		accessTokenExpiresUtc: AwareDatetime \| None = None
		accessTokenType: str \| None = None

		@model_validator(mode="before")
		@field_validator("accessTokenExpiresUtc")
		@classmethod
		def check_one_identifier(cls, values):
		if not (
		values.get("phoneNumber")
		or values.get("networkAccessIdentifier")
		or values.get("ipv4Address")
		or values.get("ipv6Address")
		def validate_expiry(cls, value: datetime \| None) -> datetime \| None:
		if value is not None and value <= datetime.now(timezone.utc):
		raise ValueError("accessTokenExpiresUtc must be in the future")
		return value

		@field_validator("accessTokenType")
		@classmethod
		def normalize_token_type(cls, value: str \| None) -> str \| None:
		return value.lower() if value is not None else None

		@model_validator(mode="after")
		def require_access_token_fields(self):
		if self.credentialType == "ACCESSTOKEN" and (
		self.accessToken is None
		or self.accessTokenExpiresUtc is None
		or self.accessTokenType is None
		):
		raise ValueError(
		"Device must include at least one of: phoneNumber, networkAccessIdentifier, ipv4Address, or ipv6Address"
		"ACCESSTOKEN requires accessToken, "
		"accessTokenExpiresUtc, and accessTokenType"
		)
		return values


		class CreateQoDSessionInput(BaseModel):
		device: DeviceInput
		qosProfile: str
		duration: int
		return self


		class CreateQoDSessionOAIInput(BaseModel):
		devicePublicAddress: str
		applicationServerIpv4Address: str
		qosProfile: str
		device: DeviceInput \| None = None
		applicationServer: ApplicationServerInput
		devicePorts: PortsSpecInput \| None = None
		applicationServerPorts: PortsSpecInput \| None = None
		qosProfile: str = Field(
		...,
		min_length=3,
		max_length=256,
		pattern=r"^[a-zA-Z0-9_.-]+$",
		)
		duration: int = Field(
		..., ge=60, le=86400, description="Duration in seconds (60-86400)"
		...,
		ge=1,
		description="Requested session duration in seconds",
		)
		sink: str \| None = Field(default=None, pattern=r"^https://.+$")
		sinkCredential: SinkCredentialInput \| None = None


		# ----- Output schema -----
		@@ -73,17 +146,25 @@ class GetQoDSessionInput(BaseModel):

		class QoDSessionFullResponse(BaseModel):
		sessionId: str
		device: dict[str, Any]
		applicationServer: dict[str, Any]
		devicePorts: dict[str, Any] \| None = None
		applicationServerPorts: dict[str, Any] \| None = None
		device: dict[str, Any] \| None = None
		applicationServer: ApplicationServerInput
		devicePorts: PortsSpecInput \| None = None
		applicationServerPorts: PortsSpecInput \| None = None
		qosProfile: str
		sink: str \| None = None
		sinkCredential: dict[str, Any] \| None = None
		duration: int
		qosStatus: str
		startedAt: str \| None = None
		expiresAt: str \| None = None
		duration: int = Field(..., ge=1)
		qosStatus: Literal["REQUESTED", "AVAILABLE", "UNAVAILABLE"]
		statusInfo: (
		Literal[
		"DURATION_EXPIRED",
		"NETWORK_TERMINATED",
		"DELETE_REQUESTED",
		]
		\| None
		) = None
		startedAt: AwareDatetime \| None = None
		expiresAt: AwareDatetime \| None = None


		class QoDSessionsList(BaseModel):

mcp_module/models/traffic_influence_models.py

0 → 100644

+56 −0

Changes for mcp_module/models/traffic_influence_models.py: 56 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from datetime import datetime, timezone
		from typing import Any, Literal
		from uuid import UUID

		from pydantic import AwareDatetime, BaseModel, Field, field_validator


		class SourceTrafficFiltersInput(BaseModel):
		sourcePort: int \| None = Field(default=None, ge=0, le=65535)


		class DestinationTrafficFiltersInput(BaseModel):
		destinationPort: int \| None = Field(default=None, ge=0, le=65535)
		destinationProtocol: str \| None = None


		class SubscriptionConfigInput(BaseModel):
		subscriptionDetail: dict[str, Any]
		subscriptionExpireTime: datetime \| None = None
		subscriptionMaxEvents: int \| None = Field(default=None, ge=1)
		initialEvent: bool \| None = None


		class AccessTokenCredentialInput(BaseModel):
		credentialType: Literal["ACCESSTOKEN"]
		accessToken: str = Field(..., min_length=1, max_length=8192, repr=False)
		accessTokenType: Literal["bearer"]
		accessTokenExpiresUtc: AwareDatetime

		@field_validator("accessTokenExpiresUtc")
		@classmethod
		def validate_expiry(cls, value: datetime) -> datetime:
		if value <= datetime.now(timezone.utc):
		raise ValueError("accessTokenExpiresUtc must be in the future")
		return value


		class SubscriptionRequestInput(BaseModel):
		protocol: Literal["HTTP", "MQTT3", "MQTT5", "AMQP", "NATS", "KAFKA"]
		sink: str = Field(..., pattern=r"^https://.+$")
		sinkCredential: AccessTokenCredentialInput \| None = None
		types: list[
		Literal["org.camaraproject.traffic-influence.v1.traffic-influence-change"]
		] = Field(..., min_length=1, max_length=1)
		config: SubscriptionConfigInput


		class CreateTrafficInfluenceInput(BaseModel):
		apiConsumerId: str
		appId: UUID
		appInstanceId: UUID \| None = None
		edgeCloudRegion: str \| None = None
		edgeCloudZoneId: UUID \| None = None
		sourceTrafficFilters: SourceTrafficFiltersInput \| None = None
		destinationTrafficFilters: DestinationTrafficFiltersInput \| None = None
		subscriptionRequest: SubscriptionRequestInput \| None = None

mcp_module/tools/qod.py

+75 −77

Changes for mcp_module/tools/qod.py: 75 added lines, 77 removed lines.
Original line number	Diff line number	Diff line
		import asyncio
		import json
		import logging
		import os
		import re
		from pathlib import Path
		from typing import Any

		import httpx
		from dotenv import load_dotenv
		@@ -24,45 +22,54 @@ 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:
		def get_qod_service_url() -> str:
		root_url = os.getenv("OEG_SERVICE_URL")
		if not root_url:
		raise OSError("OEG_SERVICE_URL environment variable is not set")
		return session_service_url

		return f"{root_url.rstrip('/')}/quality-on-demand/v1"


		def get_qod_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

		async def create_qod_session_oai(inp: CreateQoDSessionOAIInput) -> dict:
		"""
		Create QoD session using a static JSON template and return full JSON response.

		Inputs:
		- devicePublicAddress
		- applicationServerIpv4Address
		- qosProfile
		- duration
		async def create_qod_session_oai(
		inp: CreateQoDSessionOAIInput,
		xCorrelator: str \| None = None,
		) -> dict:
		"""
		template_path = root_dir / "json_templates" / "qod_create_session_request.json"
		if not template_path.exists():
		raise FileNotFoundError(f"{template_path} not found")
		Create a QoD session for a device and application server.

		payload = json.loads(
		await asyncio.to_thread(template_path.read_text, encoding="utf-8")
		Supply the QoS profile and duration in seconds.
		Optionally specify port filters and an HTTPS notification sink.
		"""
		payload = inp.model_dump(
		mode="json",
		by_alias=True,
		exclude_none=True,
		)

		# Dynamic fields requested by user.
		payload["device"]["ipv4Address"]["publicAddress"] = inp.devicePublicAddress
		payload["device"]["ipv4Address"]["privateAddress"] = inp.devicePublicAddress
		payload["applicationServer"]["ipv4Address"] = inp.applicationServerIpv4Address
		payload["qosProfile"] = inp.qosProfile
		payload["duration"] = inp.duration

		session_url = f"{get_session_service_url()}/sessions"
		logger.debug("Create QoD session payload: %s", json.dumps(payload, indent=2))
		logger.debug("Calling endpoint: %s", session_url)
		session_url = f"{get_qod_service_url()}/sessions"
		headers = get_qod_headers(xCorrelator)
		headers["content-type"] = "application/json"
		logger.debug("Creating QoD session at: %s", session_url)

		try:
		async with httpx.AsyncClient(trust_env=True) as client:
		response = await client.post(session_url, json=payload)
		response = await client.post(
		session_url,
		headers=headers,
		json=payload,
		timeout=15,
		)
		response.raise_for_status()
		except httpx.HTTPError as e:
		logger.error("Failed to create QoD session: %s", e)
		@@ -70,71 +77,62 @@ async def create_qod_session_oai(inp: CreateQoDSessionOAIInput) -> dict:

		try:
		return response.json()
		except ValueError:
		return {"status_code": response.status_code, "body": response.text}
		except ValueError as e:
		logger.error("Invalid JSON in QoD creation response: %s", e)
		raise


		async def get_qod_session_oai(inp: GetQoDSessionInput) -> QoDSessionFullResponse:
		async def get_qod_session_oai(
		inp: GetQoDSessionInput,
		xCorrelator: str \| None = None,
		) -> QoDSessionFullResponse:
		"""
		Retrieves QoS session based on session id
		Retrieve a QoD session by sessionId, including its status and expiry.
		"""

		session_url = f"{get_session_service_url()}/sessions/{inp.sessionId}"
		session_url = f"{get_qod_service_url()}/sessions/{inp.sessionId}"
		headers = get_qod_headers(xCorrelator)
		logger.debug("Fetching session details from: %s", session_url)

		try:
		async with httpx.AsyncClient(trust_env=True) as client:
		response = await client.get(session_url)
		response = await client.get(session_url, headers=headers, timeout=15)
		response.raise_for_status()
		except httpx.HTTPError as e:
		logger.error("Failed to get session %s: %s", inp.sessionId, e)
		raise

		try:
		resp_json = response.json()
		logger.debug("Session details response: %s", json.dumps(resp_json, indent=2))

		# Return all session details
		return QoDSessionFullResponse(
		sessionId=resp_json.get("sessionId"),
		device=resp_json.get("device", {}),
		applicationServer=resp_json.get("applicationServer", {}),
		devicePorts=resp_json.get("devicePorts"),
		applicationServerPorts=resp_json.get("applicationServerPorts"),
		qosProfile=resp_json.get("qosProfile"),
		sink=resp_json.get("sink"),
		sinkCredential=resp_json.get("sinkCredential"),
		duration=resp_json.get("duration"),
		qosStatus=resp_json.get("qosStatus"),
		startedAt=resp_json.get("startedAt"),
		expiresAt=resp_json.get("expiresAt"),
		)
		except ValueError as e:
		logger.error("Invalid JSON in QoD session response: %s", e)
		raise

		return QoDSessionFullResponse.model_validate(resp_json)

		async def delete_qod_session_oai(inp: GetQoDSessionInput) -> list[Any]:

		async def delete_qod_session_oai(
		inp: GetQoDSessionInput,
		xCorrelator: str \| None = None,
		) -> dict:
		"""
		Delete QoS session based on session id
		Delete a QoD session by sessionId.

		HTTP 204 indicates successful deletion.
		"""
		session_url = f"{get_session_service_url()}/sessions/{inp.sessionId}"
		logger.debug("Deleting session at: %s", session_url)
		session_url = f"{get_qod_service_url()}/sessions/{inp.sessionId}"
		headers = get_qod_headers(xCorrelator)
		logger.debug("Deleting QoD session at: %s", session_url)

		try:
		async with httpx.AsyncClient(trust_env=True) as client:
		response = await client.delete(session_url)

		# Treat success codes
		if response.status_code in (200, 204):
		# If server returns JSON list/dict, pass it through
		if response.content:
		data = response.json()
		# If server returns non-list on success, normalize to list to satisfy schema
		return data if isinstance(data, list) else [data]
		# 204 no content => empty array (NOT empty string)
		return []

		logger.warning(
		"Unexpected response deleting session %s: %s %s",
		inp.sessionId,
		response.status_code,
		response.text,
		response = await client.delete(
		session_url,
		headers=headers,
		timeout=15,
		)
		response.raise_for_status()
		return []
		except httpx.HTTPError as e:
		logger.error("Failed to delete session %s: %s", inp.sessionId, e)
		raise

		return {"status_code": response.status_code}

mcp_module/tools/traffic_influence.py

0 → 100644

+212 −0

Changes for mcp_module/tools/traffic_influence.py: 212 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import logging
		import os
		import re
		from pathlib import Path

		import httpx
		from dotenv import load_dotenv

		from mcp_module.models.traffic_influence_models import (
		CreateTrafficInfluenceInput,
		)

		logger = logging.getLogger("MCP server")

		root_dir = Path(__file__).resolve().parent.parent
		load_dotenv(dotenv_path=root_dir / ".env")


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

		return f"{root_url.rstrip('/')}/traffic-influence/v0.10"


		def get_traffic_influence_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


		async def create_traffic_influence(
		inp: CreateTrafficInfluenceInput,
		xCorrelator: str \| None = None,
		idempotencyKey: str \| None = None,
		) -> dict:
		"""
		Create a traffic influence resource for an application.

		Optionally target an instance, zone, or region and supply traffic
		filters or a notification subscription.

		Reuse the idempotency key only when retrying the same creation.
		Creation does not imply that traffic influence is already active.
		"""
		if idempotencyKey is not None and len(idempotencyKey) > 128:
		raise ValueError("idempotencyKey must not exceed 128 characters")

		payload = inp.model_dump(mode="json", exclude_none=True)
		url = f"{get_traffic_influence_service_url()}/traffic-influences"

		headers = get_traffic_influence_headers(xCorrelator)
		headers["content-type"] = "application/json"
		if idempotencyKey is not None:
		headers["Idempotency-Key"] = idempotencyKey

		logger.debug("Creating traffic influence at: %s", url)

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

		try:
		data = response.json()
		except ValueError as e:
		logger.error("Invalid JSON in traffic influence response: %s", e)
		raise

		if not isinstance(data, dict):
		raise TypeError("Expected a traffic influence object from OEG")

		return data


		async def get_traffic_influences(
		appId: str \| None = None,
		xCorrelator: str \| None = None,
		) -> dict:
		"""
		List traffic influence resources visible to the authenticated caller.

		Optionally filter by application ID.
		"""
		url = f"{get_traffic_influence_service_url()}/traffic-influences"
		headers = get_traffic_influence_headers(xCorrelator)

		params = {}
		if appId is not None:
		params["appId"] = appId

		logger.debug("Fetching traffic influences from: %s", url)

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

		try:
		data = response.json()
		except ValueError as e:
		logger.error("Invalid JSON in traffic influences response: %s", e)
		raise

		if not isinstance(data, list):
		raise TypeError("Expected a list of traffic influences from OEG")

		return {"trafficInfluences": data}


		async def get_traffic_influence(
		trafficInfluenceID: str,
		xCorrelator: str \| None = None,
		) -> dict:
		"""
		Retrieve a traffic influence resource by trafficInfluenceID.

		Returns its configuration and current state.
		"""
		url = (
		f"{get_traffic_influence_service_url()}/traffic-influences/{trafficInfluenceID}"
		)
		headers = get_traffic_influence_headers(xCorrelator)
		logger.debug("Fetching traffic influence from: %s", url)

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

		try:
		data = response.json()
		except ValueError as e:
		logger.error("Invalid JSON in traffic influence response: %s", e)
		raise

		if not isinstance(data, dict):
		raise TypeError("Expected a traffic influence object from OEG")

		return data


		async def delete_traffic_influence(
		trafficInfluenceID: str,
		xCorrelator: str \| None = None,
		) -> dict:
		"""
		Request deletion of a traffic influence resource by trafficInfluenceID.

		HTTP 202 means deletion was accepted and may still be pending.
		"""
		url = (
		f"{get_traffic_influence_service_url()}/traffic-influences/{trafficInfluenceID}"
		)
		headers = get_traffic_influence_headers(xCorrelator)
		logger.debug("Deleting traffic influence at: %s", url)

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

		return {
		"status_code": response.status_code,
		"message": "Traffic influence deletion accepted; completion is pending.",
		}