Table booking service with Agents
Introduction
This example shows how to use the uagents
library to set up a table booking scenario where a customer tries to book a table at a restaurant with the following booking specifications:
- Number of people: 3 guests
- Starting hour: 7 pm
- Duration: 2 hours
The example showcases the use of agent protocols
to facilitate communication and interaction between the two agents involved in this operation.
Supporting documentation
- Creating an agent ↗️
- Creating an interval task ↗️
- Communicating with other agents ↗️
- Register in Almanac ↗️
- Almanac Contract ↗️
- Protocols ↗️
- Agents address ↗️
The protocols
Querying table protocol
The below script defines the query_proto
protocol. This protocol allows customers to inquire about tables that meet their needs. It defines the following Data Models:
TableStatus
: defines the details of a table, including its seating capacity, available start time, and end time.QueryTableRequest
: represents the information sent by a customer, specifying the number of guests, desired start time, and reservation duration.QueryTableResponse
: represents the response sent by the restaurant, containing a list of table numbers that match the customer's request.GetTotalQueries
andTotalQueries
: these models are used for an optional feature that tracks the total number of booking inquiries received by the restaurant.
The messages flow is set as follows:
- User to Restaurant: The customer sends a
QueryTableRequest
message specifying their requirements. - Restaurant to User: The restaurant processes the request and sends a
QueryTableResponse
message containing available table numbers (or an empty list if no tables match).
Here is the full script:
from typing import List
from uagents import Context, Model, Protocol
class TableStatus(Model):
seats: int
time_start: int
time_end: int
class QueryTableRequest(Model):
guests: int
time_start: int
duration: int
class QueryTableResponse(Model):
tables: List[int]
class GetTotalQueries(Model):
pass
class TotalQueries(Model):
total_queries: int
query_proto = Protocol()
@query_proto.on_message(model=QueryTableRequest, replies=QueryTableResponse)
async def handle_query_request(ctx: Context, sender: str, msg: QueryTableRequest):
tables = {int(num): TableStatus(**status) for num, status in ctx.storage._data.items() if isinstance(num, int)}
available_tables = [
number for number, status in tables.items()
if status.seats >= msg.guests and status.time_start <= msg.time_start and status.time_end >= msg.time_start + msg.duration
]
ctx.logger.info(f"Query: {msg}. Available tables: {available_tables}.")
await ctx.send(sender, QueryTableResponse(tables=available_tables))
ctx.storage.set("total_queries", int(ctx.storage.get("total_queries") or 0) + 1)
@query_proto.on_query(model=GetTotalQueries, replies=TotalQueries)
async def handle_get_total_queries(ctx: Context, sender: str, _msg: GetTotalQueries):
await ctx.send(sender, TotalQueries(total_queries=int(ctx.storage.get("total_queries") or 0)))
Booking table protocol
The book_proto
protocol empowers customers to attempt booking a specific table based on availability. It defines the following Data Models:
BookTableRequest
: Carries information sent by the customer, including the desired table number, preferred start time, and reservation duration.BookTableResponse
: Represents the restaurant's response, indicating whether the booking was successful (table secured) or unsuccessful (table unavailable).
The messages flow is set as follows:
- User to Restaurant: The customer sends a
BookTableRequest
message specifying their chosen table and reservation details. - Restaurant to User: The restaurant checks table availability and sends a
BookTableResponse
message confirming the booking (if successful) or informing the customer about unavailability.
Here is the full script:
from uagents import Context, Model, Protocol
from .query import TableStatus
class BookTableRequest(Model):
table_number: int
time_start: int
duration: int
class BookTableResponse(Model):
success: bool
book_proto = Protocol()
@book_proto.on_message(model=BookTableRequest, replies=BookTableResponse)
async def handle_book_request(ctx: Context, sender: str, msg: BookTableRequest):
tables = {int(num): TableStatus(**status) for num, status in ctx.storage._data.items() if isinstance(num, int)}
table = tables[msg.table_number]
success = table.time_start <= msg.time_start and table.time_end >= msg.time_start + msg.duration
if success:
table.time_start = msg.time_start + msg.duration
ctx.storage.set(msg.table_number, table.dict())
await ctx.send(sender, BookTableResponse(success=success))
The agents
This example relies on two agents to manage the restaurant booking process: the restaurant
agent and the user
agent. The example flow is as below:
- The
user
agent sends a query to therestaurant
agent at regular intervals, requesting available tables. - The
restaurant
agent checks its stored table information and responds with a list of suitable tables based on theuser
request. - If available tables exist, the
user
agent attempts to book a table by sending a booking request with the chosen table number and desired time slot. - The
restaurant
agent verifies table availability for the requested time and duration. a. If successful, the booking is confirmed, and the table is marked as unavailable for that time slot. b. If unsuccessful, the customer is informed about the unavailability. - The
user
agent receives the booking confirmation or rejection message and logs the outcome.
Restaurant
The restaurant
agent manages table availability through the previously defined set of protocols (query_proto
and book_proto
). The agent stores information about each table, including its seating capacity and available time slots. The agent is able to respond to customer inquiries about available tables based on their desired criteria and processes booking requests and updates table availability accordingly.
Here is the full script:
from uagents import Agent, Context
from uagents.setup import fund_agent_if_low
from protocols.book import book_proto
from protocols.query import query_proto, TableStatus
restaurant = Agent(name="restaurant", port=8001, seed="restaurant secret phrase", endpoint=["http://127.0.0.1:8001/submit"])
fund_agent_if_low(restaurant.wallet.address())
# Include protocols and set table availability
restaurant.include(query_proto)
restaurant.include(book_proto)
TABLES = {
1: TableStatus(seats=2, time_start=16, time_end=22),
2: TableStatus(seats=4, time_start=19, time_end=21),
3: TableStatus(seats=4, time_start=17, time_end=19),
}
for number, status in TABLES.items():
restaurant._storage.set(number, status.dict())
if __name__ == "__main__":
restaurant.run()
User
The user
agent queries the restaurant for available tables that match the customer's needs based on multiple parameters including number of guests, preferred time, duration. The agent attempts to book a table if available tables are found by the restaurant
and receives confirmation or rejection from the restaurant
agent based on table availability.
Here is the full script:
from protocols.book import BookTableRequest, BookTableResponse
from protocols.query import QueryTableRequest, QueryTableResponse
from uagents import Agent, Context
from uagents.setup import fund_agent_if_low
RESTAURANT_ADDRESS = "agent1qw50wcs4nd723ya9j8mwxglnhs2kzzhh0et0yl34vr75hualsyqvqdzl990"
user = Agent(name="user", port=8000, seed="user secret phrase", endpoint=["http://127.0.0.1:8000/submit"])
fund_agent_if_low(user.wallet.address())
table_query = QueryTableRequest(guests=3, time_start=19, duration=2)
@user.on_interval(period=3.0, messages=QueryTableRequest)
async def interval(ctx: Context):
if not ctx.storage.get("completed"):
await ctx.send(RESTAURANT_ADDRESS, table_query)
@user.on_message(QueryTableResponse, replies={BookTableRequest})
async def handle_query_response(ctx: Context, sender: str, msg: QueryTableResponse):
if msg.tables:
ctx.logger.info("There is a free table, attempting to book one now")
await ctx.send(sender, BookTableRequest(table_number=msg.tables[0], time_start=table_query.time_start, duration=table_query.duration))
else:
ctx.logger.info("No free tables - nothing more to do")
ctx.storage.set("completed", True)
@user.on_message(BookTableResponse, replies=set())
async def handle_book_response(ctx: Context, _sender: str, msg: BookTableResponse):
ctx.logger.info(f"Table reservation was {'successful' if msg.success else 'UNSUCCESSFUL'}")
ctx.storage.set("completed", True)
if __name__ == "__main__":
user.run()
Expected output
Restaurant:
INFO: [restaurant]: Registering on almanac contract...
INFO: [restaurant]: Registering on almanac contract...complete
INFO: [restaurant]: Starting server on http://0.0.0.0:8001 (Press CTRL+C to quit)
INFO: [restaurant]: Query: guests=3 time_start=19 duration=2. Available tables: [2].
User:
INFO: [ user]: Registering on almanac contract...
INFO: [ user]: Registering on almanac contract...complete
INFO: [ user]: Starting server on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO: [ user]: There is a free table, attempting to book one now
INFO: [ user]: Table reservation was successful