Composable API that works for simple and complex agents
Async
Intuitive concurrency, streaming, and human in the loop
Versatile
Deploy as long-running, serverless, or durable
An agent loop you can read
Primitives for streaming, tool dispatch, and loop execution control, joined together using (mostly) plain Python.
class CustomAgent(ai.Agent): async def loop(self, context: ai.Context): # Custom event loop implementation for advanced use cases while context.keep_running(): async with ( ai.stream(context=context) as stream, ai.ToolRunner() as tr, ): # Process the LLM stream and concurrently start running # tool calls as they come async for event in ai.util.merge(stream, tr.events()): # Append the event to the history yield event if isinstance(event, ai.events.ToolEnd): # Schedule the tool call tr.schedule( context.resolve(event.tool_call) ) context.add(stream.message) context.add(tr.get_tool_message())
Only essentials
Build more AI apps with less framework.
# Low-level streaming APIasync with ai.stream(model, [ai.user_message("Hello!")]) as s: async for event in s: print(event)# High level agent-building API with tool calling and hooksasync with agent.run(model, [ai.user_message("Robot uprising?")]) as s: async for event in s: print(event)
class CustomAgent(ai.Agent): async def loop(self, context: ai.Context): # Custom event loop implementation for advanced use cases while context.keep_running(): async with ( ai.stream(context=context) as stream, ai.ToolRunner() as tr, ): # Process the LLM stream and concurrently start running # tool calls as they come async for event in ai.util.merge(stream, tr.events()): # Append the event to the history yield event if isinstance(event, ai.events.ToolEnd): # Schedule the tool call tr.schedule( context.resolve(event.tool_call) ) context.add(stream.message) context.add(tr.get_tool_message())
agent.py
# Low-level streaming APIasync with ai.stream(model, [ai.user_message("Hello!")]) as s: async for event in s: print(event)# High level agent-building API with tool calling and hooksasync with agent.run(model, [ai.user_message("Robot uprising?")]) as s: async for event in s: print(event)
