Guides
Integrations
Fetch Mobility Integration πŸš—οΈ
Walk-through πŸ“

Walk-through πŸ“

You first need to satisfy the requirements in Getting Started ↗️ section for this example.

Organize the project

Inside the src folder in the mobility-integrations directory for the project, you will need to create 2 different directories and a Python script:

cd src
mkdir agents
mkdir messages
touch main.py

These directories will contain the different files you will need to correctly run the project.

Define the scripts

Let's start by defining the needed agents for this and their respective behaviors and functions.

Agents

Within the agents folder created above, create 2 different sub-folders, one for each agent we plan to define:

mkdir ev_charger
mkdir geopy_car_parking

EV Charger

Enter the ev_charger directory previously created within the src folder.

  1. Let's start by creating a Python script for the ev_charger agent: touch ev_charger.py.

  2. We then import the needed classes and create an agent named ev_adaptor using the uagents Framework.

    import os
import uuid
 
import requests
from messages import EVRequest, KeyValue, UAgentResponse, UAgentResponseType
from uagents import Agent, Context, Protocol
from uagents.setup import fund_agent_if_low
 
EV_SEED = os.getenv("EV_SEED", "ev charger service secret phrase")
 
agent = Agent(
    name="ev_adaptor",
    seed=EV_SEED,
)
 
fund_agent_if_low(agent.wallet.address())

    This agent is a software entity designed to interact with an external service. It is being initialized with a cryptographic seed, referred to as EV_SEED which is used for secure communication and validation. To ensure that the agent can perform transactions, we need to make sure that the agent's wallet balance is enough and tops up the wallet if the balance is low by running the fund_agent_if_low() function.

  3. We need to define the code for configuring the interaction with the OpenChargeMap API, which provides information about EV charging stations.

    OPENCHARGEMAP_API_KEY = os.environ.get("OPENCHARGEMAP_API_KEY", "")
 
assert (
    OPENCHARGEMAP_API_KEY
), "OPENCHARGEMAP_API_KEY environment variable is missing from .env"
 
OPENCHARGEMAP_API_URL = "https://api.openchargemap.io/v3/poi?"
MAX_RESULTS = 100

    It retrieves the API key required for accessing the OpenChargeMap service from the environment variable OPENCHARGEMAP_API_KEY. An assert statement checks whether the API key is available. If the key is missing, it raises an assertion error and stops the script, notifying that the OPENCHARGEMAP_API_KEY environment variable is missing from the .env file. The base URL for the OpenChargeMap API, OPENCHARGEMAP_API_URL, is set to "https://api.openchargemap.io/v3/poi?". This URL will be used for making API requests. A constant MAX_RESULTS is defined with a value of 100, which might be used to limit the maximum number of results returned by API queries.

  4. We then need to define a function serving as the interface to the OpenChargeMap API to obtain data about nearby EV charging stations.

    def get_ev_chargers(latitude: float, longitude: float, miles_radius: float) -> list:
    """Return ev chargers available within given miles_readius of the latiture and longitude.
    this information is being retrieved from https://api.openchargemap.io/v3/poi? API
    """
    response = requests.get(
        url=OPENCHARGEMAP_API_URL
        + f"maxresults={MAX_RESULTS}&latitude={latitude}&longitude={longitude}&distance={miles_radius}",
        headers={"x-api-key": OPENCHARGEMAP_API_KEY},
        timeout=5,
    )
    if response.status_code == 200:
        return response.json()
    return []
 
ev_chargers_protocol = Protocol("EvChargers")

    The get_ev_chargers() function accepts three parameters: latitude, longitude, and miles_radius. These parameters define the geographical area to search for EV chargers and such function is used to retrieve information about EV chargers within a specified radius of a given latitude and longitude using the OpenChargeMap API. The function sends an HTTP GET request to the OpenChargeMap API using the requests.get() method. The URL for the API request is constructed by combining various parameters, including OPENCHARGEMAP_API_URL, MAX_RESULTS, latitude, longitude, and miles_radius. The request includes an "x-api-key" header, which provides the API key (OPENCHARGEMAP_API_KEY) for authentication and authorization. The timeout parameter is set to 5 seconds, indicating that the request should time out if it takes longer than 5 seconds to receive a response. The function checks the HTTP response status code. If the status code is 200 (indicating a successful response), the function returns the JSON content of the response, which likely contains information about EV chargers. If the status code is not 200, the function returns an empty list [].

    Finally, a protocol named ev_chargers_protocol is created using the Protocol class. This protocol is given the name EvChargers. This protocol is used to define message formats and handlers for specific types of messages related to EV chargers. It will be used to specify how the agent should handle messages of this type.

  5. We then define a message handler function, ev_chargers, which is used to handle messages of type EVRequest:

    @ev_chargers_protocol.on_message(model=EVRequest, replies=UAgentResponse)
async def ev_chargers(ctx: Context, sender: str, msg: EVRequest):
    ctx.logger.info(f"Received message from {sender}")
    try:
        ev_chargers = get_ev_chargers(msg.latitude, msg.longitude, msg.miles_radius)
        request_id = str(uuid.uuid4())
        conn_types = []
        options = []
        for idx, ev_station in enumerate(ev_chargers):
            for conn in ev_station["Connections"]:
                conn_types.append(conn["ConnectionType"]["Title"])
                conn_type_str = ", ".join(conn_types)
                option = f"""● EV charger: {ev_station['AddressInfo']['Title']} , located {round(ev_station['AddressInfo']['Distance'], 2)} miles from your location\n● Usage cost {ev_station['UsageCost']};\n● Type - {conn_type_str}"""
 
            options.append(KeyValue(key=idx, value=option))
        await ctx.send(
            sender,
            UAgentResponse(
                options=options,
                type=UAgentResponseType.SELECT_FROM_OPTIONS,
                request_id=request_id,
            ),
        )
    except Exception as exc:
        ctx.logger.error(exc)
        await ctx.send(
            sender, UAgentResponse(message=str(exc), type=UAgentResponseType.ERROR)
        )
 
agent.include(ev_chargers_protocol)

    The function is decorated with the .on_message() decorator. This specifies that the function should handle messages of type EVRequest and reply with messages of type UAgentResponse. The function takes three parameters: ctx, sender, and msg. The handler logs an informational message, using ctx.logger.info() method, indicating that it has received a message from the sender. The main functionality of the message handler is placed inside a try-except block to handle potential exceptions.

    It calls the get_ev_chargers() function, passing in the latitude, longitude, and miles_radius from the incoming msg. This function is expected to retrieve information about EV chargers in the specified geographical area.

    As a response, we get the request_id, which is an unique identifier for the request is generated using uuid.uuid4(), and the lists conn_types and options which are initialized to collect information about available EV chargers and their details. A loop iterates over the retrieved EV charger data (ev_chargers) and their connection types. For each EV charger, it constructs a detailed option string containing information such as charger title, distance, usage cost, and connection types. These option strings are appended to the options list along with a numeric key (idx) to represent each option.

    A response message of type UAgentResponse is constructed and sent back to the original sender using ctx.send() method. This response includes the collected options, specifies the response type as UAgentResponseType.SELECT_FROM_OPTIONS, and includes the request_id. If any exceptions occur during the processing of the message (e.g., an error in fetching EV charger data), an error message is logged, and an error response of type UAgentResponseType.ERROR is sent back to the sender. Finally, the message handler is included in the ev_chargers_protocol by calling .include() method. This step ensures that the agent is configured to handle messages of this type.

  6. Save the script.

The overall script should look as follows:

ev_charger.py
import os
import uuid
 
import requests
from messages import EVRequest, KeyValue, UAgentResponse, UAgentResponseType
from uagents import Agent, Context, Protocol
from uagents.setup import fund_agent_if_low
 
EV_SEED = os.getenv("EV_SEED", "ev charger service secret phrase")
 
agent = Agent(
    name="ev_adaptor",
    seed=EV_SEED,
)
 
fund_agent_if_low(agent.wallet.address())
 
OPENCHARGEMAP_API_KEY = os.environ.get("OPENCHARGEMAP_API_KEY", "")
 
assert (
    OPENCHARGEMAP_API_KEY
), "OPENCHARGEMAP_API_KEY environment variable is missing from .env"
 
OPENCHARGEMAP_API_URL = "https://api.openchargemap.io/v3/poi?"
MAX_RESULTS = 100
 
def get_ev_chargers(latitude: float, longitude: float, miles_radius: float) -> list:
    """Return ev chargers available within given miles_readius of the latiture and longitude.
    this information is being retrieved from https://api.openchargemap.io/v3/poi? API
    """
    response = requests.get(
        url=OPENCHARGEMAP_API_URL
        + f"maxresults={MAX_RESULTS}&latitude={latitude}&longitude={longitude}&distance={miles_radius}",
        headers={"x-api-key": OPENCHARGEMAP_API_KEY},
        timeout=5,
    )
    if response.status_code == 200:
        return response.json()
    return []
 
ev_chargers_protocol = Protocol("EvChargers")
 
@ev_chargers_protocol.on_message(model=EVRequest, replies=UAgentResponse)
async def ev_chargers(ctx: Context, sender: str, msg: EVRequest):
    ctx.logger.info(f"Received message from {sender}")
    try:
        ev_chargers = get_ev_chargers(msg.latitude, msg.longitude, msg.miles_radius)
        request_id = str(uuid.uuid4())
        conn_types = []
        options = []
        for idx, ev_station in enumerate(ev_chargers):
            for conn in ev_station["Connections"]:
                conn_types.append(conn["ConnectionType"]["Title"])
                conn_type_str = ", ".join(conn_types)
                option = f"""● EV charger: {ev_station['AddressInfo']['Title']} , located {round(ev_station['AddressInfo']['Distance'], 2)} miles from your location\n● Usage cost {ev_station['UsageCost']};\n● Type - {conn_type_str}"""
 
            options.append(KeyValue(key=idx, value=option))
        await ctx.send(
            sender,
            UAgentResponse(
                options=options,
                type=UAgentResponseType.SELECT_FROM_OPTIONS,
                request_id=request_id,
            ),
        )
    except Exception as exc:
        ctx.logger.error(exc)
        await ctx.send(
            sender, UAgentResponse(message=str(exc), type=UAgentResponseType.ERROR)
        )
 
agent.include(ev_chargers_protocol)

Geopy Car Parking agent

Enter the geopy_car_parking directory previously created within src.

  1. Let's first create a Python script within this directory and name it: touch geopy_car_parking.py

  2. We then need to import the necessary classes to then set the virtual environment and initialize our agent. We need to make sure it has enough funds available.

    import os
import uuid
 
import requests
from messages import GeoParkingRequest, KeyValue, UAgentResponse, UAgentResponseType
from uagents import Agent, Context, Protocol
from uagents.setup import fund_agent_if_low
 
GEOAPI_PARKING_SEED = os.getenv(
    "GEOAPI_PARKING_SEED", "geoapi parking adaptor agent secret phrase"
)
agent = Agent(name="geoapi_parking_adaptor", seed=GEOAPI_PARKING_SEED)
geoapi_parking_protocol = Protocol("Geoapi CarParking")
 
fund_agent_if_low(agent.wallet.address())
 
GEOAPI_API_KEY = os.getenv("GEOAPI_API_KEY", "")
 
assert GEOAPI_API_KEY, "GEOAPI_API_KEY environment variable is missing from .env"
 
PARKING_API_URL = "https://api.geoapify.com/v2/places?"

    Here, we begin by setting up several environment variables that are crucial for the agent's functionality. GEOAPI_PARKING_SEED is an environment variable that determines the seed phrase for the agent's wallet. If not provided in the environment, a default value, "geoapi parking adaptor agent secret phrase" is used instead. Then, an agent is created with the name geoapi_parking_adaptor and is initialized using the specified seed phrase.

    The fund_agent_if_low() function is called to ensure that the agent's wallet has sufficient funds for making transactions on the blockchain. Also, a protocol named geoapi_parking_protocol is created using the Protocol class. This protocol is given the name Geoapi CarParking.

    The agent expects to find an environment variable named GEOAPI_API_KEY. It asserts that this variable exists; otherwise, it raises an assertion error. This ensures that the necessary API key for the Geoapify service is available. The PARKING_API_URL variable is set to the base URL of the Geoapify Places API ("https://api.geoapify.com/v2/places?"). This URL will be used to make requests for parking place information.

  3. We then need to define a function to take the response from the Geoapify API and format it into a user-friendly representation:

    def format_parking_data(api_response) -> list:
    """
    By taking the response from the API, this function formats the response
    to be appropriate for displaying back to the user.
    """
    parking_data = []
    parking_name = "Unknown Parking"
    parking_capacity = ""
    for place in api_response["features"]:
        if "name" in place["properties"]:
            parking_name = place["properties"]["name"]
            address = place["properties"]["formatted"].split(",")[1::]
            parking_address = "".join(list(address))
        elif "formatted" in place["properties"]:
            parking_address = place["properties"]["formatted"]
        else:
            continue
        if "capacity" in place["properties"]["datasource"]["raw"]:
            parking_capacity = (
                f'{place["properties"]["datasource"]["raw"]["capacity"]} spaces'
            )
        elif "parking" in place["properties"]["datasource"]["raw"]:
            parking_capacity = (
                f'{place["properties"]["datasource"]["raw"]["parking"]} parking'
            )
        elif (
            "access" in place["properties"]["datasource"]["raw"]
            and place["properties"]["datasource"]["raw"]["access"] != "yes"
        ):
            continue
        parking_data.append(
            f"""● Car Parking: {parking_name} has {parking_capacity} at {parking_address}"""
        )
    return parking_data

    Here, we define a format_parking_data() function whose primary goal is to transform the raw data received from the API into a more readable and structured format for displaying it to users. It takes in an api_response. This parameter represents the raw response received from the Geoapify API. It's expected to contain parking place information. The function initializes an empty list, parking_data, to store the formatted parking information. Two variables, parking_name and parking_capacity, are initialized with default values. These values will be used if certain properties are missing in the API response.The function iterates through the features in the API response. Each feature represents a parking place. It checks if the feature contains a "name" property. If so, it assigns the parking place's name to the parking_name variable. It extracts the address from the "formatted" property and assigns it to the parking_address variable. The address is split by commas, and only elements from the second position onwards are considered. The function then looks for information about parking capacity in the "datasource" property of the feature. It checks if "capacity" or "parking" data is available and assigns it to the parking_capacity variable accordingly. Additionally, it checks if there is an "access" property and ensures that it's not equal to "yes" to filter out irrelevant data. If this condition is met, the function continues to the next feature.

    Finally, the function constructs a formatted string for each parking place using the extracted data. It includes the parking place's name, capacity (if available), and address. Each formatted string is appended to the parking_data list, using the .append() method. The function returns the parking_data list, which now contains formatted information about parking places. This function essentially takes the complex and raw data from the Geoapify API and refines it into a list of user-friendly strings, making it easier to present parking place details to end-users.

  4. We now have to define a function fetching parking data from a Geoapify Parking API:

    def get_parking_from_api(latitude, longitude, radius, max_r) -> list:
    """
    With all the user preferences, this function sends the request to the Geoapify Parking API,
    which returns the response.
    """
    try:
        response = requests.get(
            url=f"{PARKING_API_URL}categories=parking&filter=circle:{longitude},{latitude},{radius}&bias=proximity:{longitude},{latitude}&limit={max_r}&apiKey={GEOAPI_API_KEY}",
            timeout=60,
        )
        return response.json()
    except Exception as exc:
        print("Error: ", exc)
        return []

    The get_parking_from_api() function appears to be designed to fetch parking data from a Geoapify Parking API using user-specified parameters such as latitude, longitude, radius, and maximum results (max_r). Inside the function, a try block is used to make an HTTP GET request to the Geoapify Parking API using the requests.get() function. The URL for the API request is constructed based on the above specified input parameters. The URL is constructed using the following components:

    • PARKING_API_URL: This is presumably a global constant or variable that holds the base URL for the Geoapify Parking API.
    • categories=parking: It specifies that the API should retrieve parking-related data.
    • filter=circle:longitude,latitude,radius: It sets a circular geographic filter for the search.
    • bias=proximity:longitude,latitude: It biases the search to prioritize results closer to the specified coordinates.
    • limit=max_r: It limits the number of results to the specified maximum.
    • apiKey=GEOAPI_API_KEY: It includes an API key for authentication.

    The requests.get() method is used to make the API request, and the response is stored in the response variable. If the API request is successful, the function returns the JSON response by calling response.json() function. If there is an exception (e.g., a network issue or API error), it catches the exception and prints an error message along with the exception details. It then returns an empty list [].

  5. We then need to define a message handler function for handling GeoParkingRequest messages:

    @geoapi_parking_protocol.on_message(model=GeoParkingRequest, replies=UAgentResponse)
async def geoapi_parking(ctx: Context, sender: str, msg: GeoParkingRequest):
    """
    The function takes the request to search for parking in any location based on user preferences
    and returns the formatted response to TAGI.
    """
    ctx.logger.info(f"Received message from {sender}")
    try:
        radius_in_meter = msg.radius * 1609
        response = get_parking_from_api(
            msg.latitude, msg.longitude, radius_in_meter, msg.max_result
        )  # Sending user preferences to find nearby parking spaces.
        response = format_parking_data(
            response
        )  # Sending the API response to be made appropriate to users.
        request_id = str(uuid.uuid4())
        if len(response) > 1:
            option = f"""Here is the list of some Parking spaces nearby:\n"""
            idx = ""
            options = [KeyValue(key=idx, value=option)]
            for parking in response:
                option = parking
                options.append(KeyValue(key=idx, value=option))
            await ctx.send(
                sender,
                UAgentResponse(
                    options=options,
                    type=UAgentResponseType.SELECT_FROM_OPTIONS,
                    request_id=request_id,
                ),
            )  # Sending the response back to users.
        else:
            await ctx.send(
                sender,
                UAgentResponse(
                    message="No options available for this context",
                    type=UAgentResponseType.FINAL,
                    request_id=request_id,
                ),
            )
    except Exception as exc:
        ctx.logger.error(exc)
        await ctx.send(
            sender, UAgentResponse(message=str(exc), type=UAgentResponseType.ERROR)
        )
 
agent.include(geoapi_parking_protocol)

    We define a geoapi_parking() message handler function. The function is decorated with .on_message() decorator suggesting that it is triggered whenever a GeoParkingRequest is received. The function takes the following parameters: ctx, sender, and msg. It logs an informational message indicating that it has received a message from the sender using ctx.logger.info() method.

    The function wraps its main logic in a try block to handle exceptions:

    • Radius Conversion: It calculates the search radius in meters by multiplying the msg.radius value by 1609. This conversion might be needed to match the units expected by the get_parking_from_api() function.
    • API Request: It calls the get_parking_from_api() function with the user's preferences latitude, longitude, converted radius, and max result count. This function is expected to return parking data from an API and is used to retrieve parking spaces close to the user.

    The function then uses the format_parking_data() function to format the API response data into a suitable format for user presentation. Depending on the length of the formatted response data, it prepares a response to be sent back to the user. If there is more than one parking option (len(response) > 1), it constructs a list of options (KeyValue objects) and sends it as a select-from-options response to the user. This allows the user to choose from multiple parking options. If there is only one or no parking options available, it sends a suitable response message. If any exception occurs during the execution of the try block, it logs the error and sends an error message as a response to the user. Finally, we include this geoapi_parking_protocol protocol in the agent using the .include method.

  6. Save the script.

The overall script should look as follows:

geopy_car_parking.py
import os
import uuid
 
import requests
from messages import GeoParkingRequest, KeyValue, UAgentResponse, UAgentResponseType
from uagents import Agent, Context, Protocol
from uagents.setup import fund_agent_if_low
 
GEOAPI_PARKING_SEED = os.getenv(
    "GEOAPI_PARKING_SEED", "geoapi parking adaptor agent secret phrase"
)
agent = Agent(name="geoapi_parking_adaptor", seed=GEOAPI_PARKING_SEED)
geoapi_parking_protocol = Protocol("Geoapi CarParking")
 
fund_agent_if_low(agent.wallet.address())
 
GEOAPI_API_KEY = os.getenv("GEOAPI_API_KEY", "")
 
assert GEOAPI_API_KEY, "GEOAPI_API_KEY environment variable is missing from .env"
 
PARKING_API_URL = "https://api.geoapify.com/v2/places?"
 
def format_parking_data(api_response) -> list:
    """
    By taking the response from the API, this function formats the response
    to be appropriate for displaying back to the user.
    """
    parking_data = []
    parking_name = "Unknown Parking"
    parking_capacity = ""
    for place in api_response["features"]:
        if "name" in place["properties"]:
            parking_name = place["properties"]["name"]
            address = place["properties"]["formatted"].split(",")[1::]
            parking_address = "".join(list(address))
        elif "formatted" in place["properties"]:
            parking_address = place["properties"]["formatted"]
        else:
            continue
        if "capacity" in place["properties"]["datasource"]["raw"]:
            parking_capacity = (
                f'{place["properties"]["datasource"]["raw"]["capacity"]} spaces'
            )
        elif "parking" in place["properties"]["datasource"]["raw"]:
            parking_capacity = (
                f'{place["properties"]["datasource"]["raw"]["parking"]} parking'
            )
        elif (
            "access" in place["properties"]["datasource"]["raw"]
            and place["properties"]["datasource"]["raw"]["access"] != "yes"
        ):
            continue
        parking_data.append(
            f"""● Car Parking: {parking_name} has {parking_capacity} at {parking_address}"""
        )
    return parking_data
 
def get_parking_from_api(latitude, longitude, radius, max_r) -> list:
    """
    With all the user preferences, this function sends the request to the Geoapify Parking API,
    which returns the response.
    """
    try:
        response = requests.get(
            url=f"{PARKING_API_URL}categories=parking&filter=circle:{longitude},{latitude},{radius}&bias=proximity:{longitude},{latitude}&limit={max_r}&apiKey={GEOAPI_API_KEY}",
            timeout=60,
        )
        return response.json()
    except Exception as exc:
        print("Error: ", exc)
        return []
 
@geoapi_parking_protocol.on_message(model=GeoParkingRequest, replies=UAgentResponse)
async def geoapi_parking(ctx: Context, sender: str, msg: GeoParkingRequest):
    """
    The function takes the request to search for parking in any location based on user preferences
    and returns the formatted response to TAGI.
    """
    ctx.logger.info(f"Received message from {sender}")
    try:
        radius_in_meter = msg.radius * 1609
        response = get_parking_from_api(
            msg.latitude, msg.longitude, radius_in_meter, msg.max_result
        )  # Sending user preferences to find nearby parking spaces.
        response = format_parking_data(
            response
        )  # Sending the API response to be made appropriate to users.
        request_id = str(uuid.uuid4())
        if len(response) > 1:
            option = f"""Here is the list of some Parking spaces nearby:\n"""
            idx = ""
            options = [KeyValue(key=idx, value=option)]
            for parking in response:
                option = parking
                options.append(KeyValue(key=idx, value=option))
            await ctx.send(
                sender,
                UAgentResponse(
                    options=options,
                    type=UAgentResponseType.SELECT_FROM_OPTIONS,
                    request_id=request_id,
                ),
            )  # Sending the response bach to users.
        else:
            await ctx.send(
                sender,
                UAgentResponse(
                    message="No options available for this context",
                    type=UAgentResponseType.FINAL,
                    request_id=request_id,
                ),
            )
    except Exception as exc:
        ctx.logger.error(exc)
        await ctx.send(
            sender, UAgentResponse(message=str(exc), type=UAgentResponseType.ERROR)
        )
 
agent.include(geoapi_parking_protocol)

Messages

Inside the messages directory previously created, we need to define 4 Python scripts for structured data models representing the different types of messages being exchanged between our agents:

touch __init__.py
touch ev_charger.py
touch general.py
touch geoapi_parking.py

init.py

Inside this Python file, we need to define the following:

__init_.py
from .ev_charger import EVRequest
from .general import *
from .geoapi_parking import GeoParkingRequest

  • from .ev_charger import EVRequest: This line is importing the EVRequest class (or component) from a module named ev_charger. The use of . before ev_charger indicates that you're importing from a module in the same package or directory as the current module.

  • from .general import *: This line is using a wildcard * import to import all components (classes, functions, variables, etc.) from a module named general in the same package or directory.

  • from .geoapi_parking import GeoParkingRequest: This line is importing the GeoParkingRequest class from module named geoapi_parking.

ev_charger.py

We now need to define a data model class for representing requests related to Electric Vehicle (EV) charging. Inside this Python file, we first need to import the Model class from uagents library, and then define an EVRequest class. The class defines three attributes as class variables. These attributes represent the data associated with an EV charging request:

  • latitude: A floating-point number representing the latitude of the location where EV charging is requested.

  • longitude: A floating-point number representing the longitude of the location where EV charging is requested.

  • miles_radius: A floating-point number representing the search radius (in miles) to find available EV charging stations around the specified location.

    ev_charger.py
    from uagents import Model
 
class EVRequest(Model):
    """
    Represents an Electric Vehicle (EV) Charging Request.
 
    Attributes:
        latitude (float): The latitude of the location where EV charging is requested.
        longitude (float): The longitude of the location where EV charging is requested.
        miles_radius (float): The search radius (in miles) to find available EV charging stations
                             around the specified location.
    """
 
    latitude: float
    longitude: float
    miles_radius: float

general.py

We now need to define the general.py script for generalized structured data models for messages to be exchanged between our agents. We import the necessary classes and define the following classes:

general.py
rom typing import List, Optional
 
from uagents import Model
 
class UAgentResponseType(Enum):
    """
    Enumeration representing different types of responses from the User Agent.
 
    Attributes:
        ERROR (str): Represents an error response type.
        SELECT_FROM_OPTIONS (str): Represents a response type where the user needs to select from given options.
        FINAL_OPTIONS (str): Represents a response type where the user has selected the final option.
    """
 
    ERROR = "error"
    SELECT_FROM_OPTIONS = "select_from_options"
    FINAL_OPTIONS = "final_options"
 
 
class KeyValue(Model):
    """
    Represents a key-value pair.
 
    Attributes:
        key (str): The key of the pair.
        value (str): The value associated with the key.
    """
 
    key: str
    value: str
 
 
class UAgentResponse(Model):
    """
    Represents a User Agent Response.
 
    Attributes:
        type (UAgentResponseType): The type of the response.
        agent_address (str, optional): The address of the agent to which the response is directed.
        message (str, optional): A message related to the response.
        options (List[KeyValue], optional): A list of key-value pairs representing response options.
        request_id (str, optional): An optional identifier for the corresponding request.
    """
 
    type: UAgentResponseType
    agent_address: Optional[str]
    message: Optional[str]
    options: Optional[List[KeyValue]]
    request_id: Optional[str]

  • UAgentResponseType: this is an enumeration that defines different types of responses that can be generated by a user agent. It has three attributes: ERROR (Represents an error response type), SELECT_FROM_OPTIONS (Represents a response type where the user needs to select from given options), and FINAL_OPTIONS (Represents a response type where the user has selected the final option). These attributes are defined as strings and associated with their respective response types.

  • KeyValue: this is a data model class that represents a key-value pair. It has two attributes: key (a string representing the key of the pair), and value (a string representing the value associated with the key).

  • UAgentResponse: this is a data model class that represents a user Agent Response. It has several attributes: type (an attribute of type UAgentResponseType representing the type of the response), agent_address (an optional string attribute representing the address of the agent to which the response is directed), message(an optional string attribute representing a message related to the response), options (an optional list of KeyValue objects representing response options), and request_id (an optional string attribute serving as an identifier for the corresponding request).

geoapi_parking.py

Within the geoapi_parking.py, we need to define a data model class for representing requests related to parking information in a geographic location:

geoapi_parking.py
from uagents import Model
 
class GeoParkingRequest(Model):
    """
    Represents a Geo Parking Request.
 
    Attributes:
        latitude (float): The latitude of the location for which parking is requested.
        longitude (float): The longitude of the location for which parking is requested.
        radius (int): The search radius (in miles) to find parking spots around the specified location.
        max_result (int, optional): The maximum number of parking spots to be returned. Default is 5.
    """
 
    latitude: float
    longitude: float
    radius: int
    max_result: int = 5

The class defines four attributes as class variables. These attributes represent the data associated with a parking request:

  • latitude: a floating-point number representing the latitude of the location for which parking is requested.
  • longitude: a floating-point number representing the longitude of the location for which parking is requested.
  • radius: an integer representing the search radius (in miles) to find parking spots around the specified location.
  • max_result: an integer representing the maximum number of parking spots to be returned. It has a default value of 5, meaning that if not explicitly provided, the default value will be used.

Main script

Now, we are ready to define the script for our main.py Python file originally created at the beginning of this guide.

  1. In the script, let's start by importing the necessary modules and classes:

    from agents.ev_charger.ev_charger import agent as ev_charger_agent
from agents.geopy_car_parking.geopy_car_parking import agent as geopy_car_parking_agent
from uagents import Bureau
    • ev_charger_agent is imported from the agents.ev_charger.ev_charger module.
    • geopy_car_parking_agent is imported from the agents.geopy_car_parking.geopy_car_parking module.
    • Bureau is imported from the uagents library.

  2. We then would need to create an instance of the Bureau class, and configure it with the following parameters:

    • endpoint: it specifies the endpoint URL where the bureau can submit information. In this case, it's set to "http://127.0.0.1:8000/submit".
    • port: it specifies the port on which the bureau operates. In this case, it's set to 8000.
    if __name__ == "__main__":
    bureau = Bureau(endpoint="http://127.0.0.1:8000/submit", port=8000)

  3. We then need to add the two agents to the bureau:

        print(f"Adding Ev charger agent to Bureau: {ev_charger_agent.address}")
    bureau.add(ev_charger_agent)
    print(
        f"Adding geopy car parking agent to Bureau: {geopy_car_parking_agent.address}"
    )
    bureau.add(geopy_car_parking_agent)
    bureau.run()
    • ev_charger_agent is added to the bureau using bureau.add(ev_charger_agent).
    • geopy_car_parking_agent is added to the bureau using bureau.add(geopy_car_parking_agent).

  4. Save the script.

main.py
from agents.ev_charger.ev_charger import agent as ev_charger_agent
from agents.geopy_car_parking.geopy_car_parking import agent as geopy_car_parking_agent
from uagents import Bureau
 
if __name__ == "__main__":
    bureau = Bureau(endpoint="http://127.0.0.1:8000/submit", port=8000)
    print(f"Adding Ev charger agent to Bureau: {ev_charger_agent.address}")
    bureau.add(ev_charger_agent)
    print(
        f"Adding geopy car parking agent to Bureau: {geopy_car_parking_agent.address}"
    )
    bureau.add(geopy_car_parking_agent)
    bureau.run()

Run the script

We are finally ready to run the example we presented in this guide. We would need to run the main.py script to do so.

To run the project and its agents, first make sure to be in the right directory and that you activated the virtual environment:

cd src
poetry shell
python main.py

Now you have the agents up and running to perform mobility integrations using the provided APIs. Happy integrating! πŸŽ‰

Getting started 🎯References