In this tutorial, we explore how to build a small yet functional multi-agent system using the uAgents framework. We set up three agents — Directory, Seller, and Buyer — that communicate via well-defined message protocols to simulate a real-world marketplace interaction. We design message schemas, define agent behaviors, and implement request-response cycles to demonstrate discovery, negotiation, and transaction among agents, all running asynchronously in a shared event loop. Through this, we understand how autonomous agents collaborate, trade, and efficiently maintain decentralized workflows. Check out the Full Codes here.
!pip -q install "uagents>=0.11.2"
import asyncio, random
from typing import List, Dict, Optional
from uagents import Agent, Context, Bureau, Model, Protocol
class ServiceAnnounce(Model):
category: str
endpoint: str
class ServiceQuery(Model):
category: str
class ServiceList(Model):
addresses: List[str]
class OfferRequest(Model):
item: str
max_price: int
class Offer(Model):
item: str
price: int
qty: int
class Order(Model):
item: str
qty: int
class Receipt(Model):
item: str
qty: int
total: int
ok: bool
note: Optional[str] = NoneWe begin by installing the uAgents library and defining all the message models that underpin our communication system. We create structured data types for announcements, queries, offers, and orders, enabling agents to exchange information seamlessly. Check out the Full Codes here.
registry_proto = Protocol(name="registry", version="1.0")
trade_proto = Protocol(name="trade", version="1.0")
directory = Agent(name="directory", seed="dir-seed-001")
seller = Agent(name="seller", seed="seller-seed-001")
buyer = Agent(name="buyer", seed="buyer-seed-001")
directory.include(registry_proto)
seller.include(trade_proto)
buyer.include(registry_proto)
buyer.include(trade_proto)
@registry_proto.on_message(model=ServiceAnnounce)
async def on_announce(ctx: Context, sender: str, msg: ServiceAnnounce):
reg = await ctx.storage.get("reg") or {}
reg.setdefault(msg.category, set()).add(sender)
await ctx.storage.set("reg", reg)
ctx.logger.info(f"Registered {sender} under '{msg.category}'")
@registry_proto.on_message(model=ServiceQuery)
async def on_query(ctx: Context, sender: str, msg: ServiceQuery):
reg = await ctx.storage.get("reg") or {}
addrs = sorted(list(reg.get(msg.category, set())))
await ctx.send(sender, ServiceList(addresses=addrs))
ctx.logger.info(f"Returned {len(addrs)} providers for '{msg.category}'")We set up the Directory, Seller, and Buyer agents and define the registry protocol that manages service discovery. We make the directory respond to announcements and queries, allowing agents to register and locate each other dynamically. Check out the Full Codes here.
CATALOG: Dict[str, Dict[str, int]] = {
"camera": {"price": 120, "qty": 3},
"laptop": {"price": 650, "qty": 2},
"headphones": {"price": 60, "qty": 5},
}
@seller.on_event("startup")
async def seller_start(ctx: Context):
await ctx.send(directory.address, ServiceAnnounce(category="electronics", endpoint=seller.address))
ctx.logger.info("Seller announced to directory")
@trade_proto.on_message(model=OfferRequest)
async def on_offer_request(ctx: Context, sender: str, req: OfferRequest):
item = CATALOG.get(req.item)
if not item:
await ctx.send(sender, Offer(item=req.item, price=0, qty=0))
return
price = max(1, int(item["price"] * (0.9 + 0.2 * random.random())))
if price > req.max_price or item["qty"] We create the Seller agent’s catalog and implement logic for responding to offer requests and processing orders. We simulate real-world trading by adding variable pricing and stock management, showing how the seller negotiates and completes transactions. Check out the Full Codes here.
@buyer.on_event("startup")
async def buyer_start(ctx: Context):
ctx.logger.info("Buyer querying directory for electronics...")
resp = await ctx.ask(directory.address, ServiceQuery(category="electronics"), expects=ServiceList, timeout=5.0)
sellers = resp.addresses if resp else []
if not sellers:
return
target = sellers[0]
desired = "laptop"
budget = 700
ctx.logger.info(f"Requesting offer for '{desired}' within budget {budget} from {target}")
offer = await ctx.ask(target, OfferRequest(item=desired, max_price=budget), expects=Offer, timeout=5.0)
if not offer or offer.price = 1 else 0
if qty == 0:
return
ctx.logger.info(f"Placing order for {qty} x {offer.item} at {offer.price}")
receipt = await ctx.ask(target, Order(item=offer.item, qty=qty), expects=Receipt, timeout=5.0)
if receipt and receipt.ok:
ctx.logger.info(f"ORDER SUCCESS: {receipt.qty} x {receipt.item} | total={receipt.total}")We program the Buyer agent to discover sellers, request offers, and place orders based on availability and budget. We observe how the buyer interacts with the seller through asynchronous communication to complete a purchase successfully. Check out the Full Codes here.
@buyer.on_interval(period=6.0)
async def periodic_discovery(ctx: Context):
seen = await ctx.storage.get("seen") or 0
if seen >= 1:
return
await ctx.storage.set("seen", seen + 1)
ctx.logger.info("Periodic discovery tick -> re-query directory")
resp = await ctx.ask(directory.address, ServiceQuery(category="electronics"), expects=ServiceList, timeout=3.0)
n = len(resp.addresses) if resp else 0
ctx.logger.info(f"Periodic: directory reports {n} seller(s)")
bureau = Bureau()
bureau.add(directory)
bureau.add(seller)
bureau.add(buyer)
async def run_demo(seconds=10):
task = asyncio.create_task(bureau.run_async())
try:
await asyncio.sleep(seconds)
finally:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
print("\n✅ Demo run complete.\n")
try:
loop = asyncio.get_running_loop()
await run_demo(10)
except RuntimeError:
asyncio.run(run_demo(10))We add periodic discovery to have the buyer recheck available sellers, then have the Bureau run all agents together. We launch the asynchronous runtime to see the full marketplace simulation unfold and complete smoothly.
In conclusion, we have seen our agents discover one another, negotiate an offer, and complete a transaction entirely through message-based interactions. We realize how uAgents simplifies multi-agent orchestration by combining structure, communication, and state management seamlessly within Python. As we run this example, we not only witness a dynamic, autonomous system in action but also gain insight into how the same architecture can be extended to complex decentralized marketplaces, AI collaborations, and intelligent service networks, all within a lightweight, easy-to-use framework.
Check out the Full Codes here. Feel free to check out our GitHub Page for Tutorials, Codes and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is committed to harnessing the potential of Artificial Intelligence for social good. His most recent endeavor is the launch of an Artificial Intelligence Media Platform, Marktechpost, which stands out for its in-depth coverage of machine learning and deep learning news that is both technically sound and easily understandable by a wide audience. The platform boasts of over 2 million monthly views, illustrating its popularity among audiences.