Quickstart

pip install zgent

import asyncio
from zgent import (
    AgentDefinition, AgentEngine, AgentOptions, Context,
    DispatchAction, DispatchResult, Dispatcher, MemoryChannel,
)
from zgent.message import AgentResultMessage, AgentTextMessage, UserMessage

async def main():
    # Create channels for input and output
    input_ch = MemoryChannel(name="input")
    output_ch = MemoryChannel(name="output")

    # Define an agent
    definition = AgentDefinition(
        system_prompt="You are a helpful assistant. Respond briefly.",
        engine_type=AgentEngine.CLAUDE,
        dispatcher=Dispatcher(
            lambda msg: DispatchResult(key="main", action=DispatchAction.ROUTE)
        ),
        options=AgentOptions(prefix="demo"),
    )

    # Create a context with receivers and senders
    ctx = Context(
        id="hello-world",
        cwd="/tmp",
        receivers=[input_ch],
        senders=[output_ch],
        definitions=[definition],
    )

    await input_ch.connect()
    await output_ch.connect()

    # Start serving in background
    task = asyncio.create_task(ctx.serve())

    # Send a message
    await input_ch.inject(UserMessage(session_id="main", text="Say hello in one word."))

    # Collect response
    while True:
        msg = await asyncio.wait_for(output_ch.receive(), timeout=30.0)
        if isinstance(msg, AgentTextMessage):
            print(f"Agent: {msg.text}")
        if isinstance(msg, AgentResultMessage):
            break

    await ctx.stop()
    task.cancel()

asyncio.run(main())