fetch-logo
ConceptsConceptsGuidesGuidesExamplesExamplesReferencesReferencesAPIsAPIs
GitHub (opens in a new tab)
  • Examples
      • Create your first agent
      • Interval task with Agents
      • Multiple agents communication
      • Local Network Interaction
      • Agents communication
      • Agents storage
      • Table booking service with Agents
      • Cleaning service with Agents
      • Send tokens with Agents
      • Verify messages with Agents
      • Communicating with other agents wallet
      • Agents communication using Agentverse Mailbox service
      • Agents Broadcast
      • Agents Name Service
      • Query an agent using a proxy API
      • Register a dice roll agent as a Function
      • Register a coin toss agent as a Function
      • Register a local agent as a Function
      • Using News API to build network of Primary and Secondary functions
      • Locally Hosted Agent with LangChain Integration
      • Hugging face API agent as a Function
      • On Query decorator example
      • Agents and Functions Creation using APIs
      • Adding Secret to agent using Agentverse API
      • React app with agents 'on_query' decorator
      • Open Dialogue Chit-Chat
      • Predefined Dialogue Chit-Chat
      • Chat API example
      • DeltaV Dialogue Chit-Chat
  • Examples
      • Create your first agent
      • Interval task with Agents
      • Multiple agents communication
      • Local Network Interaction
      • Agents communication
      • Agents storage
      • Table booking service with Agents
      • Cleaning service with Agents
        • Introduction
      • Send tokens with Agents
      • Verify messages with Agents
      • Communicating with other agents wallet
      • Agents communication using Agentverse Mailbox service
      • Agents Broadcast
      • Agents Name Service
      • Query an agent using a proxy API
      • Register a dice roll agent as a Function
      • Register a coin toss agent as a Function
      • Register a local agent as a Function
      • Using News API to build network of Primary and Secondary functions
      • Locally Hosted Agent with LangChain Integration
      • Hugging face API agent as a Function
      • On Query decorator example
      • Agents and Functions Creation using APIs
      • Adding Secret to agent using Agentverse API
      • React app with agents 'on_query' decorator
      • Open Dialogue Chit-Chat
      • Predefined Dialogue Chit-Chat
      • Chat API example
      • DeltaV Dialogue Chit-Chat
Examples
Intermediate topics
Cleaning service with Agents

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 ann uAgents library tools.

Supporting documentation

  • Creating an agent ↗️
  • Creating an interval task ↗️
  • Communicating with other agents ↗️
  • Register in Almanac ↗️
  • Almanac Contract ↗️
  • Protocols ↗️
  • Agents address ↗️

The Protocols

init.py

This protocol acts as a bridge 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:

__init__.py
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: so to define the provider's service schedule, including maximum service distance, start and end times, and minimum hourly price.

Here is the full code:

models.py
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

cleaner.py
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

user.py
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()

Was this page helpful?

Table booking service with AgentsSend tokens with Agents
footer-logo

Main website

Integrations

Events

We’re hiring!

Twitter (opens in a new tab)Telegram (opens in a new tab)Discord (opens in a new tab)GitHub (opens in a new tab)Youtube (opens in a new tab)LinkedIn (opens in a new tab)Reddit (opens in a new tab)
Sign up for developer updates