Commit 33c2b855 authored by Natalia Grigoriadou's avatar Natalia Grigoriadou
Browse files

added location retrieval

parent 4e8e0f5c

mcp_module/app.py

+2 −0
Changes for mcp_module/app.py: 2 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -22,6 +22,7 @@ from mcp_module.tools.edge_application import ( 
    instantiate_app,

    register_app_definition,

)

from mcp_module.tools.location import retrieve_location

from mcp_module.tools.qod import (

    create_qod_session_oai,

    delete_qod_session_oai,
@@ -64,6 +65,7 @@ SAFE_TOOLS: list[Callable[..., Any]] = [ 
    get_traffic_influences,

    get_traffic_influence,

    delete_traffic_influence,

    retrieve_location,

]



for tool in SAFE_TOOLS:

mcp_module/models/location_models.py

0 → 100644
+17 −0
Changes for mcp_module/models/location_models.py: 17 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from pydantic import BaseModel, Field



from mcp_module.models.qod_models import DeviceInput





class RetrieveLocationInput(BaseModel):

    device: DeviceInput | None = None

    maxAge: int | None = Field(

        default=None,

        ge=0,

        description="Maximum acceptable age of the location in seconds",

    )

    maxSurface: int | None = Field(

        default=None,

        ge=1,

        description="Maximum acceptable location area in square meters",

    )

mcp_module/tools/location.py

0 → 100644
+81 −0
Changes for mcp_module/tools/location.py: 81 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.location_models import RetrieveLocationInput



logger = logging.getLogger("MCP server")



root_dir = Path(__file__).resolve().parent.parent

load_dotenv(dotenv_path=root_dir / ".env")





def get_location_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('/')}/location-retrieval/v0.5"





def get_location_headers(

    x_correlator: str | None = None,

) -> dict[str, str]:

    headers = {

        "accept": "application/json",

        "content-type": "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 retrieve_location(

    inp: RetrieveLocationInput,

    xCorrelator: str | None = None,

) -> dict:

    """

    Retrieve the area in which a device is located.



    Optionally specify the maximum location age in seconds and

    maximum area in square meters.



    Returns the location timestamp and a circle or polygon area.

    """

    url = f"{get_location_service_url()}/retrieve"

    headers = get_location_headers(xCorrelator)

    payload = inp.model_dump(mode="json", exclude_none=True)



    logger.debug("Retrieving device location from: %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 retrieve device location: %s", e)

        raise



    try:

        data = response.json()

    except ValueError as e:

        logger.error("Invalid JSON in location retrieval response: %s", e)

        raise



    if not isinstance(data, dict):

        raise TypeError("Expected a location object from OEG")



    return data