Cleaning service with Agents
Introduction
This file can be run on any platform supporting Python, with the necessary install permissions. This example shows how to set up a cleaning service using Agents and uAgents library tools.
Supporting documentation
- Creating an agent
- Creating an interval task
- Communicating with other agents
- Register in Almanac
- Almanac Contract
- Protocols
- Agents address
- How to use agents to send tokens
The Protocols
init.py
This protocol acts as a bridge between cleaning service providers and users requesting cleaning services.
The protocol defines the format for messages exchanged between the two parties. Users send ServiceRequest
messages specifying details about their desired cleaning service, including location, time, duration, types of services needed, and their maximum budget. Providers respond with ServiceResponse
messages indicating whether they can accept the request and, if so, their proposed price. Once a user confirms the booking with a ServiceBooking
message containing all the agreed-upon details and price, the provider sends a BookingResponse
message to confirm success or failure.
The protocol ensures the provider's services are available to the user's location. It uses the geopy
library to calculate the distance between the user and the provider. Additionally, it verifies if the requested cleaning services are offered by the provider and if the desired cleaning time falls within the provider's pre-defined availability schedule. Finally, it compares the user's budget with the provider's minimum hourly price multiplied by the cleaning duration to ensure affordability.
Here is the full code:
from datetime import datetime, timedelta
from typing import List
from geopy.distance import geodesic
from geopy.geocoders import Nominatim
from uagents import Context, Model, Protocol
from .models import Availability, Provider, User
PROTOCOL_NAME = "cleaning"
PROTOCOL_VERSION = "0.1.0"
class ServiceRequest(Model):
user: str
location: str
time_start: datetime
duration: timedelta
services: List[int]
max_price: float
class ServiceResponse(Model):
accept: bool
price: float
class ServiceBooking(Model):
location: str
time_start: datetime
duration: timedelta
services: List[int]
price: float
class BookingResponse(Model):
success: bool
cleaning_proto = Protocol(name=PROTOCOL_NAME, version=PROTOCOL_VERSION)
def in_service_region(
location: str, availability: Availability, provider: Provider
) -> bool:
geolocator = Nominatim(user_agent="micro_agents")
user_location = geolocator.geocode(location)
cleaner_location = geolocator.geocode(provider.location)
if user_location is None:
raise RuntimeError(f"user location {location} not found")
if cleaner_location is None:
raise RuntimeError(f"provider location {provider.location} not found")
cleaner_coordinates = (cleaner_location.latitude, cleaner_location.longitude)
user_coordinates = (user_location.latitude, user_location.longitude)
service_distance = geodesic(user_coordinates, cleaner_coordinates).miles
in_range = service_distance <= availability.max_distance
return in_range
@cleaning_proto.on_message(model=ServiceRequest, replies=ServiceResponse)
async def handle_query_request(ctx: Context, sender: str, msg: ServiceRequest):
provider = await Provider.filter(name=ctx.name).first()
availability = await Availability.get(provider=provider)
services = [int(service.type) for service in await provider.services]
markup = provider.markup
user, _ = await User.get_or_create(name=msg.user, address=sender)
msg_duration_hours: float = msg.duration.total_seconds() / 3600
ctx.logger.info(f"Received service request from user `{user.name}`")
if (
set(msg.services) <= set(services)
and in_service_region(msg.location, availability, provider)
and availability.time_start <= msg.time_start
and availability.time_end >= msg.time_start + msg.duration
and availability.min_hourly_price * msg_duration_hours < msg.max_price
):
accept = True
price = markup * availability.min_hourly_price * msg_duration_hours
ctx.logger.info(f"I am available! Proposing price: {price}.")
else:
accept = False
price = 0
ctx.logger.info("I am not available. Declining request.")
await ctx.send(sender, ServiceResponse(accept=accept, price=price))
@cleaning_proto.on_message(model=ServiceBooking, replies=BookingResponse)
async def handle_book_request(ctx: Context, sender: str, msg: ServiceBooking):
provider = await Provider.filter(name=ctx.name).first()
availability = await Availability.get(provider=provider)
services = [int(service.type) for service in await provider.services]
user = await User.get(address=sender)
msg_duration_hours: float = msg.duration.total_seconds() / 3600
ctx.logger.info(f"Received booking request from user `{user.name}`")
success = (
set(msg.services) <= set(services)
and availability.time_start <= msg.time_start
and availability.time_end >= msg.time_start + msg.duration
and msg.price <= availability.min_hourly_price * msg_duration_hours
)
if success:
availability.time_start = msg.time_start + msg.duration
await availability.save()
ctx.logger.info("Accepted task and updated availability.")
# send the response
await ctx.send(sender, BookingResponse(success=success))
models.py
We now need to define the data structure for the cleaning service application. We account for the following Models
:
ServiceTypes
: to represent different cleaning services (floor, window, laundry, etc.).Users
: for information like name, address, and creation time.Service
: for cleaning service type offered.Provider
: for details like name, location, creation time, and links to their availability and offered services.Availability
: to define the provider's service schedule, including maximum service distance, start and end times, and minimum hourly price.
Here is the full code:
from enum import IntEnum
from tortoise import fields, models
class ServiceType(IntEnum):
FLOOR = 1
WINDOW = 2
LAUNDRY = 3
IRON = 4
BATHROOM = 5
class User(models.Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=64)
address = fields.CharField(max_length=100)
created_at = fields.DatetimeField(auto_now_add=True)
class Service(models.Model):
id = fields.IntField(pk=True)
type = fields.IntEnumField(ServiceType)
class Provider(models.Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=64)
location = fields.CharField(max_length=64)
created_at = fields.DatetimeField(auto_now_add=True)
availability = fields.ReverseRelation["Availability"]
services = fields.ManyToManyField("models.Service")
markup = fields.FloatField(default=1.1)
class Availability(models.Model):
id = fields.IntField(pk=True)
provider = fields.OneToOneField("models.Provider", related_name="availability")
max_distance = fields.IntField(default=10)
time_start = fields.DatetimeField()
time_end = fields.DatetimeField()
min_hourly_price = fields.FloatField(default=0.0)
Agents
The Cleaner Agent
from datetime import datetime
from protocols.cleaning import cleaning_proto
from protocols.cleaning.models import Availability, Provider, Service, ServiceType
from pytz import utc
from tortoise import Tortoise
from uagents import Agent, Context
cleaner = Agent(
name="cleaner",
port=8001,
seed="cleaner secret phrase",
endpoint={
"http://127.0.0.1:8001/submit": {},
},
)
# build the cleaning service agent from the cleaning protocol
cleaner.include(cleaning_proto)
@cleaner.on_event("startup")
async def startup(_ctx: Context):
await Tortoise.init(
db_url="sqlite://db.sqlite3", modules={"models": ["protocols.cleaning.models"]}
)
await Tortoise.generate_schemas()
provider = await Provider.create(name=cleaner.name, location="London Kings Cross")
floor = await Service.create(type=ServiceType.FLOOR)
window = await Service.create(type=ServiceType.WINDOW)
laundry = await Service.create(type=ServiceType.LAUNDRY)
await provider.services.add(floor)
await provider.services.add(window)
await provider.services.add(laundry)
await Availability.create(
provider=provider,
time_start=utc.localize(datetime.fromisoformat("2022-01-31 00:00:00")),
time_end=utc.localize(datetime.fromisoformat("2023-05-01 00:00:00")),
max_distance=10,
min_hourly_price=5,
)
@cleaner.on_event("shutdown")
async def shutdown(_ctx: Context):
await Tortoise.close_connections()
if __name__ == "__main__":
cleaner.run()
User
from datetime import datetime, timedelta
from protocols.cleaning import (
BookingResponse,
ServiceBooking,
ServiceRequest,
ServiceResponse,
)
from protocols.cleaning.models import ServiceType
from pytz import utc
from uagents import Agent, Context
CLEANER_ADDRESS = (
"test-agent://agent1qdfdx6952trs028fxyug7elgcktam9f896ays6u9art4uaf75hwy2j9m87w"
)
user = Agent(
name="user",
port=8000,
seed="cleaning user recovery phrase",
endpoint={
"http://127.0.0.1:8000/submit": {},
},
)
request = ServiceRequest(
user=user.name,
location="London Kings Cross",
time_start=utc.localize(datetime.fromisoformat("2023-04-10 16:00:00")),
duration=timedelta(hours=4),
services=[ServiceType.WINDOW, ServiceType.LAUNDRY],
max_price=60,
)
MARKDOWN = 0.8
@user.on_interval(period=3.0, messages=ServiceRequest)
async def interval(ctx: Context):
ctx.storage.set("markdown", MARKDOWN)
completed = ctx.storage.get("completed")
if not completed:
ctx.logger.info(f"Requesting cleaning service: {request}")
await ctx.send(CLEANER_ADDRESS, request)
@user.on_message(ServiceResponse, replies=ServiceBooking)
async def handle_query_response(ctx: Context, sender: str, msg: ServiceResponse):
markdown = ctx.storage.get("markdown")
if msg.accept:
ctx.logger.info("Cleaner is available, attempting to book now")
booking = ServiceBooking(
location=request.location,
time_start=request.time_start,
duration=request.duration,
services=request.services,
price=markdown * msg.price,
)
await ctx.send(sender, booking)
else:
ctx.logger.info("Cleaner is not available - nothing more to do")
ctx.storage.set("completed", True)
@user.on_message(BookingResponse, replies=set())
async def handle_book_response(ctx: Context, _sender: str, msg: BookingResponse):
if msg.success:
ctx.logger.info("Booking was successful")
else:
ctx.logger.info("Booking was UNSUCCESSFUL")
ctx.storage.set("completed", True)
if __name__ == "__main__":
user.run()