Coding guide for building a fully functional multi-agent marketplace using uAgent
In this tutorial, we’ll explore how to build a small but functional multi-agent system using: uAgents frame. We set up three agents—catalog, seller, and buyer—that communicate via a well-defined messaging protocol to simulate real-world marketplace interactions. We design message patterns, define agent behavior, and implement request-response cycles to demonstrate discovery, negotiation, and transactions between agents, all running asynchronously in a shared event loop. Through this, we understand how autonomous agents collaborate, transact, and effectively maintain decentralized workflows. Check The complete code is 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 first install the uAgents library and define all the message models that support our communication system. We create structured data types for announcements, inquiries, quotes and orders to enable agents to seamlessly exchange information. Check The complete code is 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 defined the registration protocol that governs service discovery. We make the directory responsive to announcements and queries, allowing agents to dynamically register and target each other. Check The complete code is 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 a directory of seller agents and implement the logic to respond to quote requests and process orders. We simulate real-world transactions by adding variable pricing and inventory management, showing how sellers negotiate and close deals. Check The complete code is 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 buyer’s agents to discover sellers, request quotes and place orders based on availability and budget. We observe how buyers interact with sellers through asynchronous communication to successfully complete a purchase. Check The complete code is 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 added periodic discovery functionality to let buyers recheck available sellers, and then let the bureau manage all agents together. We start the asynchronous runtime to see the full market simulation unfold and complete smoothly.
In summary, we see our agents discover each other, negotiate offers, and complete transactions entirely through message-based interactions. We recognize how uAgents simplify multi-agent orchestration by seamlessly combining structure, communication, and state management in Python. When we ran this example, we not only witnessed a dynamic, autonomous system in action, but also gained insight into how the same architecture can be extended to complex decentralized markets, AI collaboration, and smart service networks, all within a lightweight, easy-to-use framework.
Check The complete code is here. Please feel free to check out our GitHub page for tutorials, code, and notebooks. In addition, welcome to follow us twitter And don’t forget to join our 100k+ ML SubReddit and subscribe our newsletter. wait! Are you using Telegram? Now you can also join us via telegram.
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 the benefit of society. His most recent endeavor is the launch of Marktechpost, an AI media platform that stands out for its in-depth coverage of machine learning and deep learning news that is technically sound and easy to understand for a broad audience. The platform has more than 2 million monthly views, which shows that it is very popular among viewers.
🙌 FOLLOW MARKTECHPOST: Add us as your go-to source on Google.
