A Coding Guide to Build a Fully Functional Multi-Agent Marketplace Using uAgent

In this tutorial, we discover how to construct a small but useful multi-agent system utilizing the uAgents framework. We arrange three brokers — Directory, Seller, and Buyer — that talk through well-defined message protocols to simulate a real-world market interplay. We design message schemas, outline agent behaviors, and implement request-response cycles to reveal discovery, negotiation, and transaction amongst brokers, all working asynchronously in a shared occasion loop. Through this, we perceive how autonomous brokers collaborate, commerce, and effectively keep decentralized workflows. Check out the Full Codes here.

Copy Code

!pip -q set up "uagents>=0.11.2"


import asyncio, random
from typing import List, Dict, Optional
from uagents import Agent, Context, Bureau, Model, Protocol


class ServiceAnnounce(Model):
   class: str
   endpoint: str


class ServiceQuery(Model):
   class: str


class ServiceList(Model):
   addresses: List[str]


class OfferRequest(Model):
   merchandise: str
   max_price: int


class Offer(Model):
   merchandise: str
   value: int
   qty: int


class Order(Model):
   merchandise: str
   qty: int


class Receipt(Model):
   merchandise: str
   qty: int
   complete: int
   okay: bool
   word: Optional[str] = None

We start by putting in the uAgents library and defining all of the message fashions that underpin our communication system. We create structured knowledge varieties for bulletins, queries, gives, and orders, enabling brokers to change data seamlessly. Check out the Full Codes here.

Copy Code

registry_proto = Protocol(identify="registry", model="1.0")
trade_proto = Protocol(identify="commerce", model="1.0")


listing = Agent(identify="listing", seed="dir-seed-001")
vendor = Agent(identify="vendor", seed="seller-seed-001")
purchaser = Agent(identify="purchaser", seed="buyer-seed-001")


listing.embody(registry_proto)
vendor.embody(trade_proto)
purchaser.embody(registry_proto)
purchaser.embody(trade_proto)


@registry_proto.on_message(mannequin=ServiceAnnounce)
async def on_announce(ctx: Context, sender: str, msg: ServiceAnnounce):
   reg = await ctx.storage.get("reg") or {}
   reg.setdefault(msg.class, set()).add(sender)
   await ctx.storage.set("reg", reg)
   ctx.logger.information(f"Registered {sender} underneath '{msg.class}'")


@registry_proto.on_message(mannequin=ServiceQuery)
async def on_query(ctx: Context, sender: str, msg: ServiceQuery):
   reg = await ctx.storage.get("reg") or {}
   addrs = sorted(checklist(reg.get(msg.class, set())))
   await ctx.ship(sender, ServiceList(addresses=addrs))
   ctx.logger.information(f"Returned {len(addrs)} suppliers for '{msg.class}'")

We arrange the Directory, Seller, and Buyer brokers and outline the registry protocol that manages service discovery. We make the listing reply to bulletins and queries, permitting brokers to register and find one another dynamically. Check out the Full Codes here.

Copy Code

CATALOG: Dict[str, Dict[str, int]] = {
   "digital camera": {"value": 120, "qty": 3},
   "laptop computer": {"value": 650, "qty": 2},
   "headphones": {"value": 60, "qty": 5},
}


@vendor.on_event("startup")
async def seller_start(ctx: Context):
   await ctx.ship(listing.tackle, ServiceAnnounce(class="electronics", endpoint=vendor.tackle))
   ctx.logger.information("Seller introduced to listing")


@trade_proto.on_message(mannequin=OfferRequest)
async def on_offer_request(ctx: Context, sender: str, req: OfferRequest):
   merchandise = CATALOG.get(req.merchandise)
   if not merchandise:
       await ctx.ship(sender, Offer(merchandise=req.merchandise, value=0, qty=0))
       return
   value = max(1, int(merchandise["price"] * (0.9 + 0.2 * random.random())))
   if value > req.max_price or merchandise["qty"] <= 0:
       await ctx.ship(sender, Offer(merchandise=req.merchandise, value=0, qty=0))
       return
   await ctx.ship(sender, Offer(merchandise=req.merchandise, value=value, qty=merchandise["qty"]))
   ctx.logger.information(f"Offered {req.merchandise} at {value} with qty {merchandise['qty']}")


@trade_proto.on_message(mannequin=Order)
async def on_order(ctx: Context, sender: str, order: Order):
   merchandise = CATALOG.get(order.merchandise)
   if not merchandise or merchandise["qty"] < order.qty:
       await ctx.ship(sender, Receipt(merchandise=order.merchandise, qty=0, complete=0, okay=False, word="Not sufficient inventory"))
       return
   complete = merchandise["price"] * order.qty
   merchandise["qty"] -= order.qty
   await ctx.ship(sender, Receipt(merchandise=order.merchandise, qty=order.qty, complete=complete, okay=True, word="Thanks!"))

We create the Seller agent’s catalog and implement logic for responding to provide requests and processing orders. We simulate real-world buying and selling by including variable pricing and inventory administration, displaying how the vendor negotiates and completes transactions. Check out the Full Codes here.

Copy Code

@purchaser.on_event("startup")
async def buyer_start(ctx: Context):
   ctx.logger.information("Buyer querying listing for electronics...")
   resp = await ctx.ask(listing.tackle, ServiceQuery(class="electronics"), expects=ServiceList, timeout=5.0)
   sellers = resp.addresses if resp else []
   if not sellers:
       return
   goal = sellers[0]
   desired = "laptop computer"
   finances = 700
   ctx.logger.information(f"Requesting provide for '{desired}' inside finances {finances} from {goal}")
   provide = await ctx.ask(goal, OfferRequest(merchandise=desired, max_price=finances), expects=Offer, timeout=5.0)
   if not provide or provide.value <= 0:
       return
   qty = 1 if provide.qty >= 1 else 0
   if qty == 0:
       return
   ctx.logger.information(f"Placing order for {qty} x {provide.merchandise} at {provide.value}")
   receipt = await ctx.ask(goal, Order(merchandise=provide.merchandise, qty=qty), expects=Receipt, timeout=5.0)
   if receipt and receipt.okay:
       ctx.logger.information(f"ORDER SUCCESS: {receipt.qty} x {receipt.merchandise} | complete={receipt.complete}")

We program the Buyer agent to uncover sellers, request gives, and place orders based mostly on availability and finances. We observe how the customer interacts with the vendor by asynchronous communication to full a buy efficiently. Check out the Full Codes here.

Copy Code

@purchaser.on_interval(interval=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.information("Periodic discovery tick -> re-query listing")
   resp = await ctx.ask(listing.tackle, ServiceQuery(class="electronics"), expects=ServiceList, timeout=3.0)
   n = len(resp.addresses) if resp else 0
   ctx.logger.information(f"Periodic: listing stories {n} vendor(s)")


bureau = Bureau()
bureau.add(listing)
bureau.add(vendor)
bureau.add(purchaser)


async def run_demo(seconds=10):
   process = asyncio.create_task(bureau.run_async())
   strive:
       await asyncio.sleep(seconds)
   lastly:
       process.cancel()
       strive:
           await process
       besides asyncio.CancelledError:
           move
   print("n Demo run full.n")


strive:
   loop = asyncio.get_running_loop()
   await run_demo(10)
besides RuntimeError:
   asyncio.run(run_demo(10))

We add periodic discovery to have the customer recheck out there sellers, then have the Bureau run all brokers collectively. We launch the asynchronous runtime to see the total market simulation unfold and full easily.

In conclusion, we now have seen our brokers uncover each other, negotiate a suggestion, and full a transaction fully by message-based interactions. We understand how uAgents simplifies multi-agent orchestration by combining construction, communication, and state administration seamlessly inside Python. As we run this instance, we not solely witness a dynamic, autonomous system in motion but in addition acquire perception into how the identical structure will be prolonged to complicated decentralized marketplaces, AI collaborations, and clever service networks, all inside a light-weight, easy-to-use framework.

Check out the Full Codes here. Feel free to take a look at our GitHub Page for Tutorials, Codes and Notebooks. Also, be happy to observe us on Twitter and don’t overlook to be a part of our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.

The put up A Coding Guide to Build a Fully Functional Multi-Agent Marketplace Using uAgent appeared first on MarkTechPost.