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.
-
Let's start by creating a Python script for the
ev_charger
agent:touch ev_charger.py
. -
We then import the needed classes and create an agent named
ev_adaptor
using theuagents
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 thefund_agent_if_low()
function. -
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 variableOPENCHARGEMAP_API_KEY
. Anassert
statement checks whether the API key is available. If the key is missing, it raises an assertion error and stops the script, notifying that theOPENCHARGEMAP_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 constantMAX_RESULTS
is defined with a value of100
, which might be used to limit the maximum number of results returned by API queries. -
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
, andmiles_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 therequests.get()
method. The URL for the API request is constructed by combining various parameters, includingOPENCHARGEMAP_API_URL
,MAX_RESULTS
,latitude
,longitude
, andmiles_radius
. The request includes an"x-api-key"
header, which provides the API key (OPENCHARGEMAP_API_KEY
) for authentication and authorization. Thetimeout
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 theProtocol
class. This protocol is given the nameEvChargers
. 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. -
We then define a message handler function,
ev_chargers
, which is used to handle messages of typeEVRequest
:@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 typeEVRequest
and reply with messages of typeUAgentResponse
. The function takes three parameters:ctx
,sender
, andmsg
. The handler logs an informational message, usingctx.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 thelatitude
,longitude
, andmiles_radius
from the incomingmsg
. 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 usinguuid.uuid4()
, and the listsconn_types
andoptions
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 theoptions
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 usingctx.send()
method. This response includes the collected options, specifies the response type asUAgentResponseType.SELECT_FROM_OPTIONS
, and includes therequest_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 typeUAgentResponseType.ERROR
is sent back to the sender. Finally, the message handler is included in theev_chargers_protocol
by calling.include()
method. This step ensures that the agent is configured to handle messages of this type. -
Save the script.
The overall script should look as follows:
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
.
-
Let's first create a Python script within this directory and name it:
touch geopy_car_parking.py
-
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 namegeoapi_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 namedgeoapi_parking_protocol
is created using theProtocol
class. This protocol is given the nameGeoapi 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. ThePARKING_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. -
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 anapi_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
andparking_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 theparking_name
variable. It extracts the address from the"formatted"
property and assigns it to theparking_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 theparking_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 theparking_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. -
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 aslatitude
,longitude
,radius
, and maximum results (max_r
). Inside the function, atry
block is used to make an HTTP GET request to the Geoapify Parking API using therequests.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[]
. -
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 aGeoParkingRequest
is received. The function takes the following parameters:ctx
,sender
, andmsg
. It logs an informational message indicating that it has received a message from the sender usingctx.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 theget_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 thisgeoapi_parking_protocol
protocol in the agent using the.include
method. - Radius Conversion: It calculates the search radius in meters by multiplying the
-
Save the script.
The overall script should look as follows:
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:
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 namedev_charger
. The use of.
beforeev_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 theGeoParkingRequest
class from module namedgeoapi_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.pyfrom 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:
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), andFINAL_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), andvalue
(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 typeUAgentResponseType
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 ofKeyValue
objects representing response options), andrequest_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:
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.
-
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 theagents.ev_charger.ev_charger
module.geopy_car_parking_agent
is imported from theagents.geopy_car_parking.geopy_car_parking
module.Bureau
is imported from theuagents
library.
-
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 to8000
.
if __name__ == "__main__": bureau = Bureau(endpoint="http://127.0.0.1:8000/submit", port=8000)
-
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 thebureau
usingbureau.add(ev_charger_agent)
.geopy_car_parking_agent
is added to thebureau
usingbureau.add(geopy_car_parking_agent)
.
-
Save the script.
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! π