الـ Agent Loop خطوة خطوة

import anthropic
from dotenv import load_dotenv
from tools_registry import TOOLS
from dispatcher import run_tool

load_dotenv()
client = anthropic.Anthropic()
SYSTEM = ("You are a customer service agent for an Egyptian online store. "
          "Use tools to check facts; never guess order details. Reply in Egyptian Arabic.")

def run_agent(user_message: str, max_steps: int = 10) -> str:
    messages = [{"role": "user", "content": user_message}]
    tools = [t["schema"] for t in TOOLS.values()]

    for step in range(max_steps):                    # hard limit: never loop forever
        response = client.messages.create(
            model="claude-sonnet-5", max_tokens=1024,
            system=SYSTEM, tools=tools, messages=messages,
        )
        if response.stop_reason != "tool_use":       # done -> return final text
            return "".join(b.text for b in response.content if b.type == "text")

        messages.append({"role": "assistant", "content": response.content})
        results = []
        for block in response.content:               # may contain SEVERAL tool calls
            if block.type == "tool_use":
                content, is_error = run_tool(block.name, block.input)
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": content, "is_error": is_error})
        messages.append({"role": "user", "content": results})   # all results together

    return "Sorry, I couldn't finish this request. A human agent will follow up."

print(run_agent("الأوردر 1024 فين؟ وكمان عايز أعرف الشحن لأسوان بكام لو طلبت بـ 640 جنيه"))

from types import SimpleNamespace as NS
from dispatcher import run_tool

# FakeClient: a scripted stand-in with the same response shape as the real API.
# It lets us test the loop without an API key. NOT a real model.
class FakeClient:
    def __init__(self, script):
        self.script, self.calls = script, 0
        self.messages = NS(create=self._create)
    def _create(self, **kwargs):
        reply = self.script[self.calls]; self.calls += 1
        return reply

def tool_use(id, name, input): return NS(type="tool_use", id=id, name=name, input=input)
def text(t): return NS(type="text", text=t)

client = FakeClient([
    # step 1: the "model" asks for TWO tools in parallel
    NS(stop_reason="tool_use", content=[
        text("Let me check both."),
        tool_use("tu_1", "get_order_status", {"order_id": "1024"}),
        tool_use("tu_2", "shipping_cost", {"governorate": "Aswan", "order_total": 640}),
    ]),
    # step 2: final answer
    NS(stop_reason="end_turn", content=[text("أوردرك خارج للتوصيل وهيوصل النهارده قبل 9، والشحن لأسوان 75 جنيه.")]),
])

messages = [{"role": "user", "content": "الأوردر 1024 فين؟ والشحن لأسوان بكام؟"}]
for step in range(10):
    response = client.messages.create(messages=messages)
    print(f"--- step {step + 1}: stop_reason = {response.stop_reason}")
    if response.stop_reason != "tool_use":
        print("FINAL:", response.content[0].text)
        break
    messages.append({"role": "assistant", "content": response.content})
    results = []
    for block in response.content:
        if block.type == "tool_use":
            content, is_error = run_tool(block.name, block.input)
            print(f"  tool {block.name}({block.input}) -> {content}")
            results.append({"type": "tool_result", "tool_use_id": block.id,
                            "content": content, "is_error": is_error})
    messages.append({"role": "user", "content": results})

print("messages in history:", len(messages))