AI SDK for 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(
                            self.resolve(event.tool_call)
                        )

                context.add(stream.message)
                context.add(tr.get_tool_message())

# Low-level streaming API
async 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 hooks
async with agent.run(model, [ai.user_message("Robot uprising?")]) as s:
    async for event in s:
        print(event)