added post traffic influence tool

Natalia Grigoriadourequested to merge
feature/traffic-influence into develop

Adds the create_traffic_influence_device MCP tool for OEG’s POST /traffic-influence/v0.10/traffic-influence-devices endpoint,

Also, updated the README file with all implemented tools.

Merge request reports

mcp_module/models/traffic_influence_models.py

+9 −0
Changes for mcp_module/models/traffic_influence_models.py: 9 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -4,6 +4,8 @@ from uuid import UUID 


from pydantic import AwareDatetime, BaseModel, Field, field_validator



from mcp_module.models.qod_models import DeviceInput





class SourceTrafficFiltersInput(BaseModel):

    sourcePort: int | None = Field(default=None, ge=0, le=65535)
@@ -59,3 +61,10 @@ class CreateTrafficInfluenceInput(BaseModel): 
    sourceTrafficFilters: SourceTrafficFiltersInput | None = None

    destinationTrafficFilters: DestinationTrafficFiltersInput | None = None

    subscriptionRequest: SubscriptionRequestInput | None = None





class CreateTrafficInfluenceDeviceInput(CreateTrafficInfluenceInput):

    device: DeviceInput | None = Field(

        default=None,

        description="The device, identified by ipv4Address. OEG requires it.",

    )

mcp_module/tools/traffic_influence.py

+52 −0
Changes for mcp_module/tools/traffic_influence.py: 52 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -3,6 +3,7 @@ import logging 
import httpx



from mcp_module.models.traffic_influence_models import (

    CreateTrafficInfluenceDeviceInput,

    CreateTrafficInfluenceInput,

)

from mcp_module.tools.oeg_errors import raise_for_oeg_status
@@ -67,6 +68,57 @@ async def create_traffic_influence( 
    return data





async def create_traffic_influence_device(

    inp: CreateTrafficInfluenceDeviceInput,

    xCorrelator: str | None = None,

) -> dict:

    """

    Create device-specific traffic influence toward local application instances.



    Identify the device by its IPv4 address. OEG cannot infer the device from

    the authorization token and refuses other identifiers.



    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.

    """

    payload = inp.model_dump(mode="json", exclude_none=True, exclude={"idempotencyKey"})

    url = f"{get_traffic_influence_service_url()}/traffic-influence-devices"



    headers = get_traffic_influence_headers(xCorrelator)

    headers["content-type"] = "application/json"

    if inp.idempotencyKey is not None:

        headers["Idempotency-Key"] = inp.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,

            )

            raise_for_oeg_status(response)

    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,

mcp_module/README.md

+15 −6
Changes for mcp_module/README.md: 15 added lines, 6 removed lines.
Original line number Diff line number Diff line
@@ -33,17 +33,26 @@ mcp_module/ 


| Tool | File | Description |

| --- | --- | --- |

| `create_qod_session_oai` | `tools/qod.py` | Creates a CAMARA QoD session |

| `get_qod_session_oai` | `tools/qod.py` | Retrieves QoD session details |

| `delete_qod_session_oai` | `tools/qod.py` | Deletes an existing QoD session |

| `get_app_definitions` | `tools/edge_application.py` | Lists available edge app definitions |

| `register_app_definition` | `tools/edge_application.py` | Registers an app definition from package type, image path, and port |

| `register_app_definition` | `tools/edge_application.py` | Registers an application manifest with the Edge Cloud Provider |

| `get_app_definition` | `tools/edge_application.py` | Retrieves one app definition by id |

| `delete_app_definition` | `tools/edge_application.py` | Deletes one app definition by id |

| `get_edge_cloud_zones` | `tools/edge_application.py` | Lists edge cloud zones |

| `instantiate_app` | `tools/edge_application.py` | Instantiates an app to one or more edge cloud zones |

| `get_edge_cloud_zones` | `tools/edge_application.py` | Lists edge cloud zones, optionally filtered by region and status |

| `instantiate_app` | `tools/edge_application.py` | Instantiates an app in a single edge cloud zone |

| `get_app_instances` | `tools/edge_application.py` | Lists instantiated/deployed applications |

| `delete_app_instance` | `tools/edge_application.py` | Deletes one app instance by id |

| `create_app_deployment` | `tools/edge_application.py` | Requests application deployment across one or more edge cloud zones |

| `get_app_deployments` | `tools/edge_application.py` | Lists application deployments |

| `delete_app_deployment` | `tools/edge_application.py` | Requests termination of an application deployment by id |

| `create_qod_session_oai` | `tools/qod.py` | Creates a CAMARA QoD session |

| `get_qod_session_oai` | `tools/qod.py` | Retrieves QoD session details |

| `delete_qod_session_oai` | `tools/qod.py` | Deletes an existing QoD session |

| `create_traffic_influence` | `tools/traffic_influence.py` | Creates traffic influence for an application |

| `create_traffic_influence_device` | `tools/traffic_influence.py` | Creates traffic influence for one device, identified by its IPv4 address |

| `get_traffic_influences` | `tools/traffic_influence.py` | Lists traffic influence resources, optionally filtered by app id |

| `get_traffic_influence` | `tools/traffic_influence.py` | Retrieves one traffic influence resource by id |

| `delete_traffic_influence` | `tools/traffic_influence.py` | Requests deletion of a traffic influence resource by id |

| `retrieve_location` | `tools/location.py` | Retrieves the location of a device |



---

mcp_module/app.py

+4 −1
Changes for mcp_module/app.py: 4 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -30,6 +30,7 @@ from mcp_module.tools.qod import ( 
)

from mcp_module.tools.traffic_influence import (

    create_traffic_influence,

    create_traffic_influence_device,

    delete_traffic_influence,

    get_traffic_influence,

    get_traffic_influences,
@@ -62,6 +63,7 @@ SAFE_TOOLS: list[Callable[..., Any]] = [ 
    create_app_deployment,

    delete_app_deployment,

    create_traffic_influence,

    create_traffic_influence_device,

    get_traffic_influences,

    get_traffic_influence,

    delete_traffic_influence,
@@ -69,7 +71,8 @@ SAFE_TOOLS: list[Callable[..., Any]] = [ 
]



for tool in SAFE_TOOLS:

    mcp.tool()(tool)

    mcp.add_tool(tool)





mcp_app = mcp.http_app(path="")

tests/mcp_module/test_tool_requests.py

+41 −2
Changes for tests/mcp_module/test_tool_requests.py: 41 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -11,7 +11,10 @@ from fastmcp.exceptions import ToolError 
from mcp_module.models import edge_application_models as em

from mcp_module.models.location_models import RetrieveLocationInput

from mcp_module.models.qod_models import CreateQoDSessionOAIInput, GetQoDSessionInput

from mcp_module.models.traffic_influence_models import CreateTrafficInfluenceInput

from mcp_module.models.traffic_influence_models import (

    CreateTrafficInfluenceDeviceInput,

    CreateTrafficInfluenceInput,

)

from mcp_module.security import authorization

from mcp_module.tools import edge_application as edge

from mcp_module.tools import location, qod
@@ -42,6 +45,7 @@ SESSION = { 
    "duration": 60,

    "devicePorts": {"ranges": [{"from": 80, "to": 90}]},

}

DEVICE_IPV4 = {"ipv4Address": {"publicAddress": "198.51.100.1", "publicPort": 443}}





@dataclass
@@ -218,6 +222,23 @@ def cases(): 
            status=201,

            idempotency=True,

        ),

        Case(

            traffic,

            "create_traffic_influence_device",

            {

                "inp": CreateTrafficInfluenceDeviceInput(

                    apiConsumerId="consumer",

                    appId=APP,

                    device=DEVICE_IPV4,

                    idempotencyKey="retry-1",

                ),

            },

            "POST",

            "/traffic-influence/v0.10/traffic-influence-devices",

            {"apiConsumerId": "consumer", "appId": APP, "device": DEVICE_IPV4},

            status=201,

            idempotency=True,

        ),

        Case(

            traffic,

            "get_traffic_influences",
@@ -382,4 +403,22 @@ def test_cases_cover_all_registered_tools(): 
        and node.target.id == "SAFE_TOOLS"

    )

    assert {node.id for node in registered} == {case.name for case in CASES}

    assert len(CASES) == 19
 
    assert len(CASES) == 20





def test_device_traffic_influence_without_idempotency_key(mock_oeg):

    body = {"apiConsumerId": "consumer", "appId": APP, "device": DEVICE_IPV4}



    def respond(request):

        assert request.url.path == "/traffic-influence/v0.10/traffic-influence-devices"

        assert json.loads(request.content) == body

        assert "idempotency-key" not in request.headers

        return httpx.Response(201, json={"trafficInfluenceID": APP, "state": "ordered"})



    mock_oeg(respond)

    result = asyncio.run(

        traffic.create_traffic_influence_device(

            CreateTrafficInfluenceDeviceInput(**body)

        )

    )

    assert result == {"trafficInfluenceID": APP, "state": "ordered"}