How To Async and Streaming

How-To: Async & Streaming

AgentMesh natively supports asynchronous event loops and streaming interfaces.

---

Asynchronous Invocation (ainvoke)

ainvoke runs non-blocking on the running asyncio event loop:

import asyncio
import agentmesh_adapter

async def main():
    app = agentmesh_adapter.compile(builder)
    result = await app.ainvoke({"messages": []})
    print("Async Result:", result)

asyncio.run(main())

---

Streaming (stream and astream)

Consume state updates as streaming generators:

# Synchronous Stream
for state_chunk in app.stream({"query": "AI research"}):
    print("State snapshot:", state_chunk)

# Asynchronous Stream
async def stream_workflow():
    async for chunk in app.astream({"query": "AI research"}):
        print("Async chunk:", chunk)

asyncio.run(stream_workflow())
```\n