chore: removed legacy tools from mcp_module

from fastmcp import FastMCP

from fastapi import FastAPI

from fastapi.responses import PlainTextResponse

#from mcp_module.tools.qod_legacy import create_qod_session, get_qod_session, delete_qod_session, list_qod_sessions

from mcp_module.tools.qod import create_qod_session_oai, get_qod_session_oai, delete_qod_session_oai

from mcp_module.tools.edge_application import get_app_definitions

from mcp_module.tools.user_phone import get_user_phone

from dotenv import load_dotenv

import os

import uvicorn

import json

import logging

import httpx

from pathlib import Path

from mcp_module.models.qod_models import *

import os

from dotenv import load_dotenv

from typing import List, Dict

# Configure logging

logging.basicConfig(level=logging.DEBUG, format="%(asctime)s [%(levelname)s] %(message)s")

logger = logging.getLogger("MCP server")

# Calculate the path to the root .env file

root_dir = Path(__file__).resolve().parent.parent  # go up one level from tools/

env_path = root_dir / ".env"

# Load the .env file

load_dotenv(dotenv_path=env_path)

#

# SESSION_SERVICE_URL = os.getenv("OEG_SERVICE_URL")

# if not SESSION_SERVICE_URL:

#     raise EnvironmentError("SESSION_SERVICE_URL environment variable is not set")

#

# async def create_qod_session(inp: CreateQoDSessionInput) -> QoDSessionMinimalResponse:

#     """

#     Create QoS session

#     """

#

#     # Load JSON template

#     BASE_DIR = Path(__file__).parent.parent  # Directory of the parent of the current directory

#     json_path = BASE_DIR / "json_templates" / "qod_request.json"

#

#     if not json_path.exists():

#         logger.error(f"Template file {json_path} not found")

#         raise FileNotFoundError(f"{json_path} not found")

#

#     with open(json_path, "r", encoding="utf-8") as f:

#         template = json.load(f)

#

#     # Merge inputs into the template

#     device_data = inp.device.model_dump(exclude_unset=True)

#     template["device"].update(device_data)

#     template["qosProfile"] = inp.qosProfile

#     template["duration"] = inp.duration

#

#     logger.debug("Final request payload: %s", json.dumps(template, indent=2))

#

#     try:

#         async with httpx.AsyncClient(trust_env=True) as client:

#             response = await client.post(f"{SESSION_SERVICE_URL}/sessions", json=template)

#             response.raise_for_status()

#     except httpx.HTTPError as e:

#         logger.error("Failed to send request to session service: %s", e)

#         raise

#

#     resp_json = response.json()

#     logger.debug("Raw response: %s", json.dumps(resp_json, indent=2))

#

#     # Extract only needed fields

#     return QoDSessionMinimalResponse(

#         sessionId=resp_json.get("sessionId"),

#         qosStatus=resp_json.get("qosStatus")

#     )

#

#

# async def get_qod_session(inp: GetQoDSessionInput) -> QoDSessionFullResponse:

#     """

#     Retrieves QoS session based on session id

#     """

#

#     session_url = f"{SESSION_SERVICE_URL}/sessions/{inp.sessionId}"

#     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.raise_for_status()

#     except httpx.HTTPError as e:

#         logger.error("Failed to get session %s: %s", inp.sessionId, e)

#         raise

#

#     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")

#     )

#

# async def delete_qod_session(inp: GetQoDSessionInput) -> str:

#     """

#     Delete QoS session by sessionId.

#     """

#     session_url = f"{SESSION_SERVICE_URL}/sessions/{inp.sessionId}"

#     logger.debug("Deleting session at: %s", session_url)

#

#     try:

#         async with httpx.AsyncClient(trust_env=True) as client:

#             response = await client.delete(session_url)

#         if response.status_code == 204:

#             logger.info("Session %s deleted successfully", inp.sessionId)

#             return "success"

#         else:

#             logger.warning(

#                 "Unexpected response deleting session %s: %s %s",

#                 inp.sessionId,

#                 response.status_code,

#                 response.text

#             )

#             response.raise_for_status()

#     except httpx.HTTPError as e:

#         logger.error("Failed to delete session %s: %s", inp.sessionId, e)

#         raise

#

#

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

#

#     """

#        Retrieves QoD information based on device information such as phone number

#     """

#

#     sessions_url = f"{SESSION_SERVICE_URL}/retrieve-sessions"

#     payload = inp.model_dump(exclude_unset=True)

#

#     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))

#

#     try:

#         async with httpx.AsyncClient(trust_env=True) as client:

#             response = await client.post(sessions_url, json=payload)

#             response.raise_for_status()

#     except httpx.HTTPError as e:

#         logger.error("Failed to retrieve QoS sessions: %s", e)

#         raise

#

#     resp_json = response.json()

#     logger.debug("List of sessions response: %s", json.dumps(resp_json, indent=2))

#

#     # 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

import logging

from mcp_module.models.user_models import UserPhoneInput, UserPhoneResponse

# Configure logging

logging.basicConfig(level=logging.DEBUG, format="%(asctime)s [%(levelname)s] %(message)s")

logger = logging.getLogger("MCP server")

# Dummy data for user -> phone lookup

USER_PHONE_BOOK = {

    "Carmen Canalez": "+34551234567",

    "George Papadopoulos": "+30698654321",

}

async def get_user_phone(inp: UserPhoneInput) -> UserPhoneResponse:

    """

    Return a phone number for a specific user from a lookup table.

    """

    key = inp.user

    phone = USER_PHONE_BOOK.get(key)

    if not phone:

        logger.warning("User '%s' not found in phone book", inp.user)

        raise ValueError(f"Unknown user: {inp.user}")

    return UserPhoneResponse(phoneNumber=phone)