C.W.K.
Stream
Lesson 06 of 07 · published

The Multi-Turn Tool Loop Without SDK

~22 min · raw-tool-loop, no-sdk

Level 0Tokenizer
0 XP0/54 lessons0/10 achievements
0/120 XP to next level120 XP to go0% complete

Here's the complete implementation: send request → parse streaming → detect function_call → execute → send result → repeat.

The same loop, your wiring

The conceptual loop is identical to the SDK version: build request → POST → parse response → if tool_calls execute and append → POST again → else return text. The difference is that every step is yours to instrument. You can log the raw request body, the raw response body, the parsed tool_call decisions, the handler return values — at any point, with no SDK helper hiding the shape.

This level of control is overkill for typical product code; it's exactly right when building an adapter that has to support multiple providers, when reproducing a customer bug from a captured wire trace, or when integrating with an LLM gateway that expects you to behave like a well-behaved HTTP client.

Code

Raw tool loop end to end·python
import os, json, httpx, asyncio

TOOLS_MAP = {"get_weather": get_weather}
URL = "https://api.openai.com/v1/responses"
HEADERS = {
    "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
    "Content-Type": "application/json",
}

async def agent_loop(user_message: str, max_iterations: int = 10):
    input_items = [{"role": "user", "content": user_message}]

    async with httpx.AsyncClient(timeout=None) as client:
        for i in range(max_iterations):
            body = {
                "model": "gpt-4.1",
                "input": input_items,
                "tools": tool_schemas,
                "stream": True,
            }
            text = ""
            tool_calls = []
            event_lines = []

            async with client.stream("POST", URL, headers=HEADERS, json=body) as resp:
                resp.raise_for_status()
                async for line in resp.aiter_lines():
                    if line == "":
                        if not event_lines: continue
                        event_type, data_json = None, None
                        for l in event_lines:
                            if l.startswith("event:"): event_type = l[6:].strip()
                            elif l.startswith("data:"): data_json = json.loads(l[5:].strip())
                        event_lines.clear()
                        if event_type == "response.output_text.delta":
                            text += data_json["delta"]
                            print(data_json["delta"], end="", flush=True)
                        elif event_type == "response.function_call_arguments.done":
                            tool_calls.append(data_json)
                    else:
                        event_lines.append(line)

            if not tool_calls:
                print()
                return text

            # Execute tools and add results
            for tc in tool_calls:
                result = TOOLS_MAP[tc["name"]](**json.loads(tc["arguments"]))
                input_items.append({"type": "function_call_output",
                    "call_id": tc["call_id"], "output": json.dumps(result)})

asyncio.run(agent_loop("What's the weather in Tokyo?"))

External links

Exercise

Reimplement the Track 5 tool loop entirely in raw httpx. Same 3 fake tools, same iteration cap. Add one feature the SDK can't easily give you: a per-iteration breakpoint hook(state) that lets you inspect or mutate state mid-loop.

Progress

Progress is local-only — sign in to sync across devices.
Spotted a bug or have feedback on this page?Report an Issue

Comments 0

🔔 Reply notifications (sign in)
Sign inPlease sign in to comment.

No comments yet — be the first.