היום בוובינר דיברנו על Claude Managed Agents וראינו הדגמה של הקוד ואיך ליצור את ההדגמה הזו אצלכם. אלה עיקרי הדברים שהיו שם.

1. מה זה Managed Agents

גם אנתרופיק וגם OpenAI השיקו לאחרונה מוצר של סוכנים מנוהלים. מוצרים אלה מתחרים ב Agent Core של AWS וההסבר לכל הקטגוריה הוא פשוט - להריץ סוכנים זה מסובך. בין השאר:

1. צריך לשמור את כל השיחות של כל הסוכנים.
2. צריך דרך לעצור שיחה כשדברים לא הולכים בכיוון שרציתם.
3. כשיש עומס צריך יותר מכונות, כשאין מבקרים אפשר לצמצם.
4. צריך להיות מסוגלים לחזור לשיחות ישנות גם אחרי הרבה זמן.
5. אם יש הפעלת כלים צריך להפריד בין הלקוחות ולא לאפשר ללקוח אחד להשתלט או להגיע למידע של לקוחות אחרים.

ההגיון של סוכנים מנוהלים הוא בדיוק ההגיון של הענן. אתם מביאים את התוכנה, אנחנו נביא את התשתית. אנתרופיק ו OpenAI כבר מחזיקות תשתית מטורפת להרצת סוכנים אז למה לא למכור גם אותה יחד עם הטוקנים.

2. איך לכתוב סוכן מנוהל באנתרופיק

מוצר הסוכנים המנוהלים של אנתרופיק וותיק יותר ולכן דיברנו עליו. בעמוד הזה אפשר למצוא את המדריך של אנתרופיק להתחלה מהירה:

https://platform.claude.com/docs/en/managed-agents/quickstart

אנחנו בוובינר לקחנו משם את הפקודה להרצה אינטרקטיבית של המדריך מתוך קלוד קוד:

/claude-api managed-agents-onboard

קלוד קוד שאל אותי מה אני רוצה שהסוכן יעשה ואני ביקשתי סוכן של רשימת קניות שתומך בשתי פעולות, הוספת פריט לרשימה וצפיה ברשימה. נכון רשימת קניות אמיתית גם צריכה לאפשר לסמן שקנינו משהו או למחוק פריטים אבל את זה נשאיר לגרסה 2. אחרי קצת חשיבה קלוד קוד יצר שלושה קבצים: קובץ requirements.txt עם כל התלויות, קובץ app.py עם שרת API וקובץ setup.py שמקים את הסוכן באנתרופיק.

מפתח ה API שלי היה שמור בקובץ .env ובאיטרציה הראשונה קלוד קוד שכח לטעון אותו אבל אחרי תזכורת הקוד התחיל לעבוד. הקובץ הראשון setup.py מכיל את הקוד הבא:

# ONE-TIME SETUP - run once (`python setup.py`), then copy the printed lines into .env
#
# Creates the three reusable, versioned resources the runtime app needs:
#   - an environment (the cloud sandbox)
#   - a memory store (persists the shopping list across sessions)
#   - an agent (model/system/tools - references nothing session-specific)
#
# Agents/environments could also be defined as YAML and applied with the `ant` CLI
# for CI-managed configs (see shared/anthropic-cli.md) - this script also creates
# the memory store, which the CLI doesn't have a documented syntax for yet, so
# everything is grouped here for a single one-time run.
import anthropic
from dotenv import load_dotenv

load_dotenv()

client = anthropic.Anthropic()

SHOPPING_LIST_PATH = "/shopping_list.md"


def main():
    environment = client.beta.environments.create(
        name="shopping-list-env",
        config={"type": "cloud", "networking": {"type": "unrestricted"}},
    )

    store = client.beta.memory_stores.create(
        name="Shopping List",
        description="The user's shopping list: a single markdown checklist file.",
    )
    client.beta.memory_stores.memories.create(
        store.id,
        path=SHOPPING_LIST_PATH,
        content="# Shopping List\n",
    )

    agent = client.beta.agents.create(
        name="Shopping List Agent",
        model="claude-sonnet-5",
        system=(
            "You manage the user's shopping list. It lives as a single file, "
            f"shopping_list.md, inside your attached memory store - a markdown "
            "checklist with one item per line, formatted as `- item`.\n\n"
            "When asked to add an item: read the file, add the new line only if an "
            "equivalent item isn't already present (merge quantities when given, e.g. "
            "adding 'milk' when 'milk' is already there should not create a duplicate "
            "line), then write the file back. Reply with a short confirmation of what "
            "changed.\n\n"
            "When asked for the current list: read the file and reply with the items, "
            "one per line, no extra commentary. If the list is empty, say so explicitly."
        ),
        tools=[
            {
                "type": "agent_toolset_20260401",
                "default_config": {"enabled": True},
                "configs": [
                    {"name": "web_search", "enabled": False},
                    {"name": "web_fetch", "enabled": False},
                ],
            }
        ],
    )

    print("Add these lines to .env:")
    print(f"ANTHROPIC_ENV_ID={environment.id}")
    print(f"ANTHROPIC_AGENT_ID={agent.id}")
    print(f"ANTHROPIC_AGENT_VERSION={agent.version}")
    print(f"ANTHROPIC_MEMORY_STORE_ID={store.id}")


if __name__ == "__main__":
    main()

את הקוד הזה מריצים פעם אחת ואנחנו רואים שהוא מכיל את רכיבי התשתית השונים שנוצרים:

1. סביבה - איפה הסוכן ירוץ, איזה הרשאות יהיו לו.
2. זכרון - איזה קבצים הסוכן יחזיק, במקרה שלנו קובץ markdown של רשימת הקניות.
3. סוכן - מודל, הוראות וכלים.

קלוד קוד במקור יצר את הסוכן עם מודל אופוס אני שיניתי אותו לסונט כדי לקבל תשובות מהירות יותר וזולות יותר.

הקובץ השני שנוצר הוא app.py, זה פשוט שרת API ב flask רגיל לגמרי שמתחבר לסוכן:

# RUNTIME - every invocation. Requires setup.py to have been run once and its
# printed IDs added to .env.
import os
import time

import anthropic
from dotenv import load_dotenv
from flask import Flask, jsonify, render_template, request

load_dotenv()

client = anthropic.Anthropic()

AGENT_ID = os.environ["ANTHROPIC_AGENT_ID"]
AGENT_VERSION = os.environ["ANTHROPIC_AGENT_VERSION"]
ENV_ID = os.environ["ANTHROPIC_ENV_ID"]
STORE_ID = os.environ["ANTHROPIC_MEMORY_STORE_ID"]

MEMORY_INSTRUCTIONS = (
    "This store holds one file, shopping_list.md - a markdown checklist, one item "
    "per line as `- item`. Read it before answering and write it back after any change."
)

app = Flask(__name__)


def run_single_turn(message: str) -> str:
    """Create a short-lived session, send one message, return the agent's reply text."""
    session = client.beta.sessions.create(
        agent={"type": "agent", "id": AGENT_ID, "version": AGENT_VERSION},
        environment_id=ENV_ID,
        resources=[
            {
                "type": "memory_store",
                "memory_store_id": STORE_ID,
                "access": "read_write",
                "instructions": MEMORY_INSTRUCTIONS,
            }
        ],
    )
    print(f"session: https://platform.claude.com/workspaces/default/sessions/{session.id}", flush=True)

    reply_parts: list[str] = []
    with client.beta.sessions.events.stream(session_id=session.id) as stream:
        client.beta.sessions.events.send(
            session_id=session.id,
            events=[{"type": "user.message", "content": [{"type": "text", "text": message}]}],
        )
        for event in stream:
            if event.type == "agent.message":
                for block in event.content:
                    if block.type == "text":
                        reply_parts.append(block.text)
            elif event.type == "session.status_terminated":
                break
            elif event.type == "session.status_idle":
                if event.stop_reason.type == "requires_action":
                    continue  # waiting on a tool confirmation/result we don't send here
                break  # end_turn, retries_exhausted, or budget_reached

    # This session is one-shot - nothing else will use it, so archive it instead
    # of leaving it sitting `idle` forever. The stream's `status_idle` event can
    # arrive slightly before the session's queryable status catches up, so a
    # fresh archive call can 400 with "cannot archive while running" - retry briefly.
    for attempt in range(3):
        try:
            client.beta.sessions.archive(session_id=session.id)
            break
        except anthropic.APIStatusError as e:
            if e.status_code == 400 and attempt < 2:
                time.sleep(1)
                continue
            print(f"warning: could not archive session {session.id}: {e}", flush=True)

    return "".join(reply_parts).strip()


@app.get("/")
def index():
    return render_template("index.html")


@app.get("/shopping-list")
def get_list():
    reply = run_single_turn(
        "List everything currently on the shopping list. Reply with one item per "
        "line, each line formatted as `- item`, and nothing else. If it's empty, "
        "just say so in plain text instead."
    )
    items = [
        line.strip()[1:].strip()
        for line in reply.splitlines()
        if line.strip().startswith("-")
    ]
    return jsonify({"items": items, "raw": reply})


@app.post("/shopping-list")
def add_item():
    item = (request.get_json(force=True) or {}).get("item", "").strip()
    if not item:
        return jsonify({"error": "missing 'item'"}), 400
    reply = run_single_turn(f"Add '{item}' to the shopping list.")
    return jsonify({"result": reply})


if __name__ == "__main__":
    app.run(port=8000, debug=True)

הפונקציה המעניינת כאן היא run_single_turn והיא פונה לסוכן עם בקשה ומחזירה את התשובה. הפניה לסוכן מבוצעת דרך קריאת הספריה client.beta.sessions.create כדי ליצור session ולאחריה client.beta.sessions.events.stream כדי לשלוח הודעה.

היתרון של שיטת העבודה הזו הוא הנוחות - אנחנו לא צריכים בסיס נתונים, לא צריכים לחשוב על scale ומה יקרה אם כמה לקוחות יפנו במקביל, לא צריכים לטפל בחיבורי רשת יוצאים וחיפוש, הכל מנוהל ברמת הפלטפורמה. כל השיחות מוקלטות ושמורות על השרתים של אנתרופיק וראינו שאנחנו יכולים לגשת לכל המידע דרך ממשק הניהול שלהם.

החסרונות הם המחיר והתלות - כשהסוכן רץ על התשתית של אנתרופיק אני "נעול" על המערכת שלהם וצריך לעדכן את הקוד אם וכאשר הם ישנו את הממשק ואני לא יכול להחליט מחר לשלב מודלים של OpenAI או Ollama. לגבי המחיר בדיוק כמו עם הענן אנחנו משלמים פרמיה על הגמישות.

סך הכל מבחינת נוחות עבודה Claude Managed Agents לדעתי יותר נוח מ Agent Core, במיוחד אם ממילא אתם רוצים לעבוד רק עם הסוכנים שלהם.