MCP tool for QoS information retrieval based on device identifier (1013fe99) · Commits · OOP / code / ai2

MCP_module/Dummy_backend/controllers/experimental/qod_controller.py

+21 −7

Original line number	Diff line number	Diff line
		@@ -120,7 +120,9 @@ def delete_session(sessionId: str) -> tuple:
		def retrieve_sessions(body: Dict[str, Any]) -> tuple:
		"""
		POST /retrieve-sessions
		Retrieve QoS sessions associated with a specific device
		Retrieve QoS sessions associated with a specific device.
		If the user specifies only some device identifiers (e.g., phoneNumber),
		return only those identifiers in the response.
		"""
		try:
		device = body.get("device", {})
		@@ -133,7 +135,7 @@ def retrieve_sessions(body: Dict[str, Any]) -> tuple:

		phone_number = device.get("phoneNumber")
		network_id = device.get("networkAccessIdentifier")
		ipv4_address = device.get("ipv4Address", {}).get("publicAddress")
		ipv4_address = device.get("ipv4Address", {}).get("publicAddress") if device.get("ipv4Address") else None
		ipv6_address = device.get("ipv6Address")

		# Search for sessions that match any device identifier
		@@ -141,12 +143,24 @@ def retrieve_sessions(body: Dict[str, Any]) -> tuple:
		for session in sessions_db.values():
		session_device = session.get("device", {})
		if (
		session_device.get("phoneNumber") == phone_number
		or session_device.get("networkAccessIdentifier") == network_id
		or session_device.get("ipv4Address", {}).get("publicAddress") == ipv4_address
		or session_device.get("ipv6Address") == ipv6_address
		(phone_number and session_device.get("phoneNumber") == phone_number)
		or (network_id and session_device.get("networkAccessIdentifier") == network_id)
		or (ipv4_address and session_device.get("ipv4Address", {}).get("publicAddress") == ipv4_address)
		or (ipv6_address and session_device.get("ipv6Address") == ipv6_address)
		):
		# Only include the fields that were present in the request
		filtered_device = {}
		if "phoneNumber" in device:
		filtered_device["phoneNumber"] = session_device.get("phoneNumber")
		if "networkAccessIdentifier" in device:
		filtered_device["networkAccessIdentifier"] = session_device.get("networkAccessIdentifier")
		if "ipv4Address" in device:
		filtered_device["ipv4Address"] = session_device.get("ipv4Address")
		if "ipv6Address" in device:
		filtered_device["ipv6Address"] = session_device.get("ipv6Address")

		matched_sessions.append({
		"device": filtered_device,
		"applicationServer": session["applicationServer"],
		"qosProfile": session["qosProfile"],
		"sink": session["sink"],

MCP_module/Dummy_backend/openapi.yaml

+30 −1

Original line number	Diff line number	Diff line
		@@ -207,6 +207,27 @@ paths:
		items:
		type: object
		properties:
		device:
		type: object
		properties:
		phoneNumber:
		type: string
		example: "+123456789"
		networkAccessIdentifier:
		type: string
		example: "123456789@domain.com"
		ipv4Address:
		type: object
		properties:
		publicAddress:
		type: string
		example: "203.0.113.0"
		publicPort:
		type: integer
		example: 59765
		ipv6Address:
		type: string
		example: "2001:db8:85a3:8d3:1319:8a2e:370:7344"
		applicationServer:
		type: object
		qosProfile:
		@@ -228,7 +249,14 @@ paths:
		type: string
		enum: [ REQUESTED, AVAILABLE, UNAVAILABLE ]
		example:
		- applicationServer: { }
		- device:
		phoneNumber: "+123456789"
		networkAccessIdentifier: "123456789@domain.com"
		ipv4Address:
		publicAddress: "203.0.113.0"
		publicPort: 59765
		ipv6Address: "2001:db8:85a3:8d3:1319:8a2e:370:7344"
		applicationServer: { }
		qosProfile: "QOS_L"
		sink: "https://application-server.com/notifications"
		sessionId: "3fa85f64-5717-4562-b3fc-2c963f66afa6"
		@@ -261,3 +289,4 @@ paths:
		code: "INVALID_ARGUMENT"
		message: "Client specified an invalid argument"

MCP_module/models/qod_models.py

+11 −1

Original line number	Diff line number	Diff line
		@@ -51,3 +51,13 @@ class QoDSessionFullResponse(BaseModel):

		class QoDSessionsList(BaseModel):
		device: DeviceInput

		class QoDSessionListResponse(BaseModel):
		applicationServer: Dict[str, Any]
		device: Dict[str, Any]
		qosProfile: str
		sink: Optional[str] = None
		duration: int
		qosStatus: str
		startedAt: Optional[str] = None
		expiresAt: Optional[str] = None
		No newline at end of file

MCP_module/tools/qod.py

+30 −18

Original line number	Diff line number	Diff line
		@@ -121,15 +121,16 @@ async def delete_qod_session(inp: GetQoDSessionInput) -> str:
		logger.error("Failed to delete session %s: %s", inp.sessionId, e)


		async def list_qod_sessions(inp: DeviceInput) -> List[Dict]:
		async def list_qod_sessions(inp: QoDSessionsList) -> List[QoDSessionListResponse]:

		"""
		Retrieves all QoS session details based on user device input such as phone number
		Retrieves QoD information based on device information such as phone number
		"""

		sessions_url = f"{SESSION_SERVICE_URL}/retrieve-sessions"
		payload = {"device": inp.model_dump(exclude_unset=True)}
		payload = inp.model_dump(exclude_unset=True)

		if not payload["device"]:
		if not payload.get("device"):
		raise ValueError("Device input must include at least one identifier")

		logger.debug("Fetching QoS sessions with payload: %s", json.dumps(payload, indent=2))
		@@ -144,19 +145,30 @@ async def list_qod_sessions(inp: DeviceInput) -> List[Dict]:
		resp_json = response.json()
		logger.debug("List of sessions response: %s", json.dumps(resp_json, indent=2))

		# Return the entire list of session objects
		sessions = [
		{
		"applicationServer": session.get("applicationServer"),
		"duration": session.get("duration"),
		"expiresAt": session.get("expiresAt"),
		"qosProfile": session.get("qosProfile"),
		"qosStatus": session.get("qosStatus"),
		"sessionId": session.get("sessionId"),
		"sink": session.get("sink"),
		"startedAt": session.get("startedAt")
		}
		for session in resp_json
		]
		# Make sure it's a list
		if not isinstance(resp_json, list):
		logger.error(f"Unexpected response type: {type(resp_json)}")
		return []

		sessions = []
		requested_device_fields = inp.device.model_dump(exclude_unset=True)

		for session in resp_json:
		# Include only the fields from the request
		session_device_full = session.get("device", {})
		filtered_device = {k: session_device_full.get(k) for k in requested_device_fields.keys()}

		sessions.append(QoDSessionListResponse(
		device=session.get("device", inp.device),
		applicationServer=session.get("applicationServer", {}),
		qosProfile=session.get("qosProfile", ""),
		sink=session.get("sink"),
		duration=session.get("duration", 0),
		qosStatus=session.get("qosStatus", ""),
		startedAt=session.get("startedAt"),
		expiresAt=session.get("expiresAt"),
		))

		return sessions