Aea Plugin. May 2026

def __init__(self, agent_name: str = "plugin_agent"): self.agent = Agent(name=agent_name, seed="plugin_seed_123") self._register_handlers()

def _register_handlers(self): @self.agent.on_message(model=PluginMessage) async def handle_plugin(ctx: Context, sender: str, msg: PluginMessage): ctx.logger.info(f"Plugin received: {msg.command}") if msg.command == "process_data": result = self.process(msg.data) await ctx.send(sender, PluginResponse(status="ok", result=result)) elif msg.command == "status": await ctx.send(sender, PluginResponse(status="running", result={"version": "1.0"})) else: await ctx.send(sender, PluginResponse(status="error", result={"msg": "Unknown command"})) aea plugin.

def run(self): self.agent.run() if name == " main ": plugin = MyPlugin() plugin.run() def __init__(self, agent_name: str = "plugin_agent"): self

def process(self, data: dict) -> dict: # Your plugin logic here return {"processed": True, "original": data} result={"version": "1.0"})) else: await ctx.send(sender