Tools, MCP ו-Governance — structured output, חיבורים חיצוניים ושערים דטרמיניסטיים

1. למה structured output הוא לא מותרות — הוא ההבדל בין harness אמין ל-harness שמתפרק

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

// הגישה השבירה — מצא את המספר אחרי הנקודתיים

text = response.content[0].text
price = float(text.split("price:")[1].strip().split()[0])  # 🙏 שיעבוד

זה עובד עד שזה לא. ברגע שהמודל מחזיר "The price is approximately $4.50" במקום "price: 4.50" — ה-split נשבר, ה-harness זורק IndexError, והריצה מתה באמצע. וזה לא edge case נדיר: המחקר שעליו נשען הקורס מסמן את "parsing של free-form text outputs במקום אכיפת JSON schemas" כאחת משלוש הטעויות הקלאסיות של מתחילים בבניית harness ביתי — לצד frameworks כבדים והתעלמות מ-tool latency. הסיבה לכך היא מבנית: מודלי שפה הם סטוכסטיים, והם משנים את הפורמט שלהם ב-5% מהמקרים גם כשהם מתבקשים במדויק. 5% מתוך 200 קריאות = 10 קריסות. ה-harness שלך לא יחיה ככה לאורך זמן.

הנגד הוא לא "prompt טוב יותר שמבקש יפה JSON". הנגד הוא אכיפה ברמת ה-API: לבקש מהמודל לעמוד ב-schema קשיח, ולקבל החזרה אובייקט שאתה יודע שתקין — או שגיאת validation מפורשת שאתה יכול לתקן עליה. שני המנגנונים שעושים את זה ב-Claude API הם structured outputs (אכיפת פורמט התשובה) ו-strict tool use (אכיפת ה-schema של ה-tool input).

1.1 input schema מול output schema — אוכפים את שניהם

ל-tool יש שני גבולות שאפשר לאכוף עליהם, וטעות נפוצה היא לאכוף רק אחד מהם:

• input schema — מה המודל שולח אל ה-tool. כאן strict: true מבטיח ש-tool.input תקין לפני שאתה מריץ. בלי זה, המודל יכול לשלוח {"cId": "..."} במקום {"customer_id": "..."} וה-handler שלך יקרוס.
• output schema — מה אתה מחזיר אל המודל, או מה המודל מחזיר בתשובה הסופית. כאן structured output מבטיח שהתשובה הסופית ניתנת לפרסור דטרמיניסטי.

החלוקה הזו חשובה כי ההחלטה איפה לאכוף schema תלויה במה שאתה בונה. אם ה-tool שלך קורא ל-API חיצוני ומחזיר נתונים — צריך input schema (כדי שהמודל ישלח פרמטרים תקינים) וגם output schema (כדי שתדע בדיוק איזה shape מגיע בחזרה). אם ה-tool שלך מבצע side-effect (כתיבה, מחיקה, שליחה) — input schema חיוני לבדיקה, ו-output schema פחות קריטי כי אתה מסתכל על סטטוס, לא על תוכן.

נתחיל מ-input. ככה מגדירים tool עם strict input schema ב-Claude API (Anthropic SDK, Python). שים לב לשלושת הדברים שהופכים אותו ל-strict: "strict": true, "additionalProperties": false, ו-"required":

tools/structured_tool.py

import anthropic

client = anthropic.Anthropic()  # קורא ANTHROPIC_API_KEY מה-env

GET_INVENTORY_TOOL = {
    "name": "get_inventory",
    "description": (
        "Look up current stock for a SKU. "
        "Call this whenever the user asks about availability or stock levels."
    ),
    "strict": True,  # ← אכיפת ה-input schema
    "input_schema": {
        "type": "object",
        "properties": {
            "sku": {"type": "string", "description": "Product SKU, e.g. 'TS-1024'"},
            "warehouse": {
                "type": "string",
                "enum": ["EU", "US", "IL"],  # רק שלושה ערכים חוקיים
            },
        },
        "required": ["sku", "warehouse"],
        "additionalProperties": False,  # נדרש ל-strict
    },
}

response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=16000,
    tools=[GET_INVENTORY_TOOL],
    messages=[{"role": "user", "content": "כמה TS-1024 יש במחסן באירופה?"}],
)

# בזכות strict, אנחנו יודעים ש-tool.input תקין — אין צורך ב-defensive parsing
for block in response.content:
    if block.type == "tool_use":
        sku = block.input["sku"]            # מובטח להיות string
        warehouse = block.input["warehouse"]  # מובטח להיות EU/US/IL
        print(sku, warehouse)

עכשיו ה-input. אבל מה עם ה-תשובה — המקרה שבו אתה רוצה שהמודל יחזיר אובייקט מובנה (לא יקרא ל-tool, אלא יחזיר נתונים)? כאן נכנס output_config.format:

1.2 structured output על התשובה — הדרך המומלצת עם messages.parse()

הגישה הנקייה ביותר ב-Python היא client.messages.parse() עם מודל Pydantic. הוא מייצר את ה-schema אוטומטית, מאכף אותו ב-API, ומחזיר אובייקט מטופס ומאומת — בלי ש-json.loads ידני יופיע בקוד שלך:

tools/structured_output.py

from pydantic import BaseModel
import anthropic

client = anthropic.Anthropic()

class StockResult(BaseModel):
    sku: str
    available: int
    warehouse: str
    backorder: bool

response = client.messages.parse(
    model="claude-opus-4-8",
    max_tokens=2048,
    messages=[{
        "role": "user",
        "content": "TS-1024: 42 units in EU, none on backorder. Structure it.",
    }],
    output_format=StockResult,   # ה-SDK מאכף את ה-schema ומחזיר אובייקט מאומת
)

result = response.parsed_output   # StockResult — לא מחרוזת, לא dict
print(result.available, result.backorder)  # 42 False

ואם אתה עובד ישירות מול messages.create() (לדוגמה כשאתה לא ב-Python, או רוצה שליטה מלאה ב-schema), הצורה הקנונית היא output_config.format עם json_schema גולמי:

response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=2048,
    messages=[{"role": "user", "content": "..."}],
    output_config={
        "format": {
            "type": "json_schema",
            "schema": {
                "type": "object",
                "properties": {
                    "sku": {"type": "string"},
                    "available": {"type": "integer"},
                    "backorder": {"type": "boolean"},
                },
                "required": ["sku", "available", "backorder"],
                "additionalProperties": False,
            },
        }
    },
)
import json
data = json.loads(next(b.text for b in response.content if b.type == "text"))

1.3 למה schema validation מנצח "מודל שני שבודק"

שאלה שעולה הרבה: "למה לא לתת ל-LLM שני לבדוק את ה-output של הראשון?" התשובה היא עלות, latency ודטרמיניזם. בדיקת schema היא קריאת קוד — מיקרו-שניות, אפס tokens, ותשובה בוליאנית חד-משמעית. LLM שני הוא קריאת API נוספת — מאות מילי-שניות, אלפי tokens, ותשובה שיפוטית שעלולה להשתנות בין ריצות. כשהשאלה היא "האם זה JSON תקין עם השדות הנכונים" — זו שאלה מכנית, וכלי מכני עונה עליה מהר וזול יותר מכל מודל.

המספרים הקשים: schema validation רץ ב-~0.2ms על 1KB JSON. קריאת LLM שנייה לוקחת 800ms-3000ms ועולה 200-2000 tokens (תלוי באורך). ב-batch של 1000 קריאות, ההבדל הוא 0.2 שניות לעומת 25-50 דקות — ובעלות של אפס לעומת 5-15 דולר. תשלום בכסף ובזמן על שאלה שקוד עונה עליה מיידית הוא בזבוז מובנה.

הכלל המנחה: השתמש בקוד לבדיקות מכניות (פורמט, type, טווח), ושמור LLM רק לבדיקות שיפוטיות אמיתיות (האם הסיכום נאמן למקור? האם הטון מתאים?). ערבוב של השניים — לתת ל-LLM לבדוק פורמט — הוא בזבוז, ולתת לקוד regex לשפוט איכות — הוא שביר. כל כלי לתפקיד שלו.

attempt 0 → INVALID: ValidationError: 'available' is a required property
attempt 1 → INVALID: JSONDecodeError: Unterminated string starting at: line 1 column 47
attempt 2 → VALID
{'sku': 'TS-1024', 'available': 42, 'in_stock': True}

2. retry דטרמיניסטי על schema violation — לתקן, לא לקרוס

structured output מבטיח הרבה, אבל לא הכל. שלושה מקרים עדיין יכולים להחזיר לך output לא-תקין: (א) המודל הגיע ל-max_tokens וה-JSON נחתך באמצע; (ב) אתה מאמת מול schema עשיר יותר ממה ש-Claude מאכף בצד שלו (לדוגמה אילוצים עסקיים — "available חייב להיות >= 0"); (ג) המודל החזיר JSON תקין אבל הוא ריק ({}) או לא מכיל את המידע המבוקש (זה hallucination קלאסי ל-schema — "הצורה נכונה, התוכן חסר"). במקרים האלה אתה לא רוצה לקרוס — אתה רוצה להחזיר למודל את שגיאת ה-validation ולבקש תיקון.

הנקודה הקריטית: זה retry דטרמיניסטי, לא "LLM שני שמחליט אם התשובה מספיק טובה". הטריגר הוא בדיקת קוד בוליאנית — תקין/לא תקין. ה-LLM היחיד בלולאה הוא זה שמתקן את ה-output שלו עצמו, בתגובה לשגיאה מפורשת. זו ההבחנה המהותית: אותו מודל מתקן, לא מודל אחר שופך.

הנה לולאת ה-repair המלאה. שים לב שאנחנו מצמידים תקרת max_repairs — בלי זה, מודל שעקשני בלהחזיר output שבור יכול לשרוף את כל ה-budget:

harness/retry.py

import json
import anthropic
from jsonschema import validate, ValidationError  # pip install jsonschema

client = anthropic.Anthropic()

def call_with_repair(messages, schema, *, max_repairs=2):
    """קורא למודל ומבקש output לפי schema; מתקן דטרמיניסטית עד max_repairs."""
    convo = list(messages)
    for attempt in range(max_repairs + 1):
        response = client.messages.create(
            model="claude-opus-4-8",
            max_tokens=4096,
            messages=convo,
            output_config={"format": {"type": "json_schema", "schema": schema}},
        )

        # max_tokens חתך את ה-JSON? זה כשל לתיקון, לא תשובה.
        text = next((b.text for b in response.content if b.type == "text"), "")
        try:
            data = json.loads(text)
            validate(instance=data, schema=schema)   # אכיפה בצד שלך
            return data                               # ✅ תקין — סיימנו
        except (json.JSONDecodeError, ValidationError) as err:
            if attempt == max_repairs:
                raise RuntimeError(f"schema repair failed after {max_repairs}: {err}")
            # מחזירים את ה-output השבור + השגיאה, ומבקשים תיקון
            convo = convo + [
                {"role": "assistant", "content": text},
                {"role": "user", "content": (
                    f"Your output failed validation:\n{err}\n"
                    "Return ONLY corrected JSON matching the schema."
                )},
            ]
    # לא אמור להגיע לכאן
    raise RuntimeError("unreachable")

2.1 ה-backoff הזעיר שעושה הבדל גדול

הסקריפט למעלה עושה retry מיידי. ברוב המקרים זה מספיק, אבל יש מצב שבו הוא לא: כשהכשל הוא transient (rate limit, network glitch, 503 מה-API). במקרים האלה, retry מיידי פשוט מייצר עוד כשל. הפתרון הוא exponential backoff עם jitter — מחכים זמן שגדל בכל ניסיון, עם רנדומיזציה קטנה כדי למנוע thundering herd:

import asyncio, random

async def call_with_repair_resilient(messages, schema, *, max_repairs=2):
    convo = list(messages)
    for attempt in range(max_repairs + 1):
        try:
            # ... אותו logic של קריאה + validation + repair feedback
            ...
            return data
        except (anthropic.RateLimitError, anthropic.APIConnectionError) as e:
            if attempt == max_repairs:
                raise
            # exponential backoff: 1s, 2s, 4s + jitter של 0-500ms
            wait = (2 ** attempt) + random.uniform(0, 0.5)
            await asyncio.sleep(wait)
        except (json.JSONDecodeError, ValidationError) as e:
            # ... תיקון schema
            ...

ההבדל בין retry מיידי לבין backoff: בעומס רגעי (לדוגמה spike של משתמשים), backoff מפזר את העומס מחדש ומונע "גל שני" של קרישות. עלות הריצה הבודדת: עד 4 שניות המתנה. עלות חוסר ה-backoff: כשל cascade על כל ה-batch כשה-API רץ לאט. הכלל: כל retry על תקלת רשת → backoff. כל retry על schema violation → מיידי (אין סיבה לחכות).

3. Model Context Protocol (MCP) — לחבר מקורות חיצוניים בלי integration ייעודי לכל אחד

עד עכשיו ה-tools שלך היו פונקציות Python שכתבת ידנית. זה מצוין ל-3 tools. אבל ברגע שאתה רוצה גישה ל-filesystem אמיתי, ל-DB, ל-GitHub, ל-Slack — אתה לא רוצה לכתוב integration נפרד לכל אחד מהם. כאן נכנס MCP — Model Context Protocol, סטנדרט פתוח שמתאר איך LLM מתחבר למקורות חיצוניים. כתבת MCP server פעם אחת, וכל harness שמדבר MCP יכול להשתמש בו.

הרעיון הוא הפשטה של transport: במקום שכל harness יכיר את ה-API של כל שירות, יש פרוטוקול אחיד שכל הצדדים מדברים. התוצאה היא ecosystem: יש כבר עשרות MCP servers קיימים ל-GitHub, Postgres, Slack, Notion, Google Drive, Linear, ועוד. הם נכתבו פעם אחת, והם עובדים בכל harness שמדבר MCP — בלי שאתה כותב שורה של integration code.

3.1 שתי דרכים לחבר MCP — בחר לפי איפה ה-server רץ

ב-Claude API יש שתי דרכים מובחנות, וחשוב לבחור נכון:

נתחיל מ-MCP connector — הקצר ביותר, כשיש server מרוחק. שים לב לשני החלקים שחייבים לבוא יחד: mcp_servers מצהיר על ה-server, ו-mcp_toolset ב-tools מפנה אליו בשם. השמטת ה-toolset נדחית כשגיאת validation:

harness/mcp_connector.py

import anthropic

client = anthropic.Anthropic()

response = client.beta.messages.create(
    model="claude-opus-4-8",
    max_tokens=4096,
    betas=["mcp-client-2025-11-20"],
    mcp_servers=[
        {"type": "url", "name": "company-docs",
         "url": "https://mcp.example.com/sse"},
    ],
    tools=[
        # חובה: לכל server ב-mcp_servers חייב להיות mcp_toolset שמפנה אליו בשם
        {"type": "mcp_toolset", "mcp_server_name": "company-docs"},
    ],
    messages=[{"role": "user", "content": "מצא את מדיניות ההחזרות במסמכים"}],
)

ValidationError: mcp_server 'company-docs' declared in mcp_servers
but no mcp_toolset references it in tools[]

3.2 ה-deliverable: חיבור MCP filesystem לוקלי דרך ה-harness

עכשיו הגרסה שאנחנו בונים — server לוקלי (filesystem) דרך stdio, מחובר ל-harness כ-client. הדפוס: מריצים את ה-MCP server כ-subprocess, מבקשים ממנו את רשימת ה-tools, וממירים אותם ל-tools שה-tool_runner יודע להריץ. ה-tool_runner סוגר את הלולאה אוטומטית — קורא למודל, מריץ את ה-tool מול ה-MCP server, ומזין את התוצאה חזרה:

harness/mcp_client.py

import asyncio
from anthropic import AsyncAnthropic
from anthropic.lib.tools.mcp import async_mcp_tool  # pip install "anthropic[mcp]"
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters

client = AsyncAnthropic()

async def run_with_filesystem_mcp(user_goal: str):
    # מריצים את ה-MCP filesystem server כ-subprocess, מוגבל ל-./workspace בלבד
    server = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "./workspace"],
    )
    async with stdio_client(server) as (read, write):
        async with ClientSession(read, write) as mcp:
            await mcp.initialize()

            # שואלים את ה-server אילו tools הוא חושף (read_file, write_file, ...)
            tools_result = await mcp.list_tools()

            # tool_runner סוגר את הלולאה: model → tool → result → repeat
            runner = client.beta.messages.tool_runner(
                model="claude-opus-4-8",
                max_tokens=8192,
                messages=[{"role": "user", "content": user_goal}],
                tools=[async_mcp_tool(t, mcp) for t in tools_result.tools],
            )
            async for message in runner:
                for block in message.content:
                    if block.type == "text":
                        print(block.text)

asyncio.run(run_with_filesystem_mcp(
    "קרא את workspace/notes.md וסכם אותו בשורה אחת"
))

3.3 למה MCP חוסך לך זמן — מתמטיקה פשוטה של integrations

בלי MCP, חיבור של N מקורות חיצוניים ל-M harnesses דורש כתיבה של N×M integrations ייעודיים — כל harness כותב adapter משלו לכל מקור. עם MCP, כל מקור חושף server אחד (N servers), וכל harness מדבר את הפרוטוקול פעם אחת (M clients) — סה"כ N+M במקום N×M. זו בדיוק הסיבה שה-ecosystem מסביב MCP צמח כל כך מהר: server של GitHub, של Slack, של Postgres — נכתבו פעם אחת, וכל harness שמדבר MCP נהנה מהם בחינם.

בשבילך, ה-vibe coder שבונה harness ביתי, זה אומר דבר אחד פרקטי: לפני שאתה כותב custom tool שמתחבר לשירות חיצוני — בדוק אם כבר קיים MCP server לשירות הזה. אם כן, חברת אותו ב-5 שורות במקום לכתוב ולתחזק integration שלם. אם לא, ושאתה צריך אותו בכמה harnesses — שווה לכתוב MCP server (ולא custom tool) כדי שיהיה ניתן לשימוש חוזר.

הציור האסימפטוטי מרשים: עם 20 harnesses ו-30 שירותים, בלי MCP יש לך 600 integrations לתחזק (גיהינום). עם MCP יש לך 30 servers + 20 client setups = 50 רכיבים. הבדל של פקטור 12.

[mcp] connecting to filesystem server at ./workspace ...
[mcp] tools exposed: ['read_file', 'write_file', 'list_directory', ...]
[runner] model requested tool_use: read_file(path="workspace/notes.md")
[mcp] result: "# Project notes\n- status: planning\n- deadline: 2026-Q3\n..."
[runner] model final answer: "The notes file describes a project in planning with a 2026 Q3 deadline."

4. Governance — לגדר כל קריאת tool מאחורי policy gate דטרמיניסטי

עכשיו לחלק הכי קריטי. נתת לסוכן גישה ל-bash דרך MCP filesystem. הסוכן יכול עכשיו להריץ פקודות. מה עוצר אותו מלהריץ rm -rf / או DROP TABLE users אם הוא hallucinate, או אם prompt injection במסמך שהוא קרא שכנע אותו לעשות זאת?

התשובה חייבת להיות דטרמיניסטית. לא "ה-system prompt מבקש ממנו יפה לא לעשות נזק" — מספיק hallucination אחד והבקשה המנומסת לא שווה כלום. השער חייב להיות קוד שבודק את קריאת ה-tool לפני שהיא רצה, לפי policy declarative. זה הרעיון של governance-as-code, ובמערכת אמיתית הוא ממומש על ידי כלי כמו Faramesh — execution control plane פתוח שאוכף gates דטרמיניסטיים על קריאות tool בלי LLM שני שיפוטי. אנחנו נבנה גרסה דקה של אותו עיקרון בעצמנו, כדי שתבין מה קורה מתחת למכסה.

4.1 שלוש ההחלטות של gate: allow / deny / require-human-approval

כל gate מחזיר אחת משלוש החלטות, ולכל אחת יש מקום ברור:

נכתוב policy declarative ב-YAML (זה ה-"FPL הדק" שלנו), ואחריו את ה-engine שאוכף אותו. הרעיון: ה-policy הוא נתונים, לא קוד אד-הוק — ניתן ל-audit, ל-version control, ולשינוי בלי לגעת בלולאה:

governance/policy.yaml

# policy declarative — governance-as-code. ה-prompt לא מעורב בהחלטה.
rules:
  - tool: bash
    deny_if_input_matches: '(rm\s+-rf|mkfs|dd\s+if=|:\(\)\s*\{)'  # פקודות הרסניות
    reason: "destructive shell command"

  - tool: run_sql
    deny_if_input_matches: '(?i)\b(DROP|TRUNCATE|DELETE\s+FROM\s+\w+\s*;?\s*$)\b'
    reason: "destructive SQL"

  - tool: send_email
    require_approval: true     # לגיטימי אבל רגיש — human-in-the-loop
    reason: "outbound email needs human sign-off"

  - tool: git
    deny_if_input_matches: 'push\s+.*--force'
    reason: "force-push is irreversible"

default: allow   # כל מה שלא מכוסה — מותר (read-only ברובו)

governance/policy.py

import re, yaml
from dataclasses import dataclass

@dataclass
class Decision:
    action: str          # "allow" | "deny" | "approve"
    reason: str = ""
    rule: str = ""        # איזה כלל גרם להחלטה (ל-debug)

class PolicyGate:
    """gate דטרמיניסטי — בודק כל קריאת tool לפי policy.yaml. אין LLM כאן."""

    def __init__(self, policy_path: str):
        with open(policy_path) as f:
            self.policy = yaml.safe_load(f)

    def check(self, tool_name: str, tool_input: dict) -> Decision:
        # מסרקים את ה-input פעם אחת לטקסט לצורך התאמת regex
        blob = " ".join(str(v) for v in tool_input.values())
        for rule in self.policy.get("rules", []):
            if rule["tool"] != tool_name:
                continue
            pat = rule.get("deny_if_input_matches")
            if pat and re.search(pat, blob):
                return Decision("deny", rule.get("reason", "policy violation"),
                                rule_name=f"{rule['tool']}:deny")
            if rule.get("require_approval"):
                return Decision("approve", rule.get("reason", "needs approval"),
                                rule_name=f"{rule['tool']}:approve")
        return Decision(self.policy.get("default", "allow"))

4.2 לחבר את ה-gate ללולאה — לפני כל ביצוע

ה-gate שווה רק אם הוא יושב בדיוק בין הרגע שהמודל מבקש tool לבין הרגע שאתה מריץ אותו. הנה ההשתלבות בלולאה הידנית (מהפרק הקודם), עם שלוש ההחלטות:

harness/governed_loop.py

from governance.policy import PolicyGate

gate = PolicyGate("governance/policy.yaml")

def request_human_approval(tool_name, tool_input, reason) -> bool:
    # ב-capstone זה יהפוך ל-UI / Slack; כאן — אישור בטרמינל
    print(f"⚠️  approval needed for {tool_name}: {reason}")
    print(f"    input: {tool_input}")
    return input("    allow? [y/N] ").strip().lower() == "y"

def execute_tool_call(block):
    """מיירט כל tool_use דרך ה-gate לפני ביצוע."""
    decision = gate.check(block.name, block.input)

    if decision.action == "deny":
        # מחזירים tool_result עם is_error — המודל רואה שנחסם וינסה דרך אחרת
        return {"type": "tool_result", "tool_use_id": block.id, "is_error": True,
                "content": f"BLOCKED by policy: {decision.reason}"}

    if decision.action == "approve":
        if not request_human_approval(block.name, block.input, decision.reason):
            return {"type": "tool_result", "tool_use_id": block.id, "is_error": True,
                    "content": "DENIED by human reviewer"}

    # allow (או approved) — מריצים בפועל
    result = run_tool(block.name, block.input)   # ה-handler שלך
    return {"type": "tool_result", "tool_use_id": block.id, "content": result}

4.3 human-in-the-loop כעמוד שדרה — דפוס ה-lead-enrichment

ה-require_approval הוא לא רק "אמצעי זהירות" — הוא ה-pattern שמאפשר ל-capstone לעבוד. שקול lead-enrichment pipeline: הסוכן מנטר GitHub stars על repos, מעשיר נתוני lead דרך Apollo, ומנסח טיוטת outreach email. הצעד האחרון — שליחת ה-email — חייב לעצור לאישור אנושי. זו בדיוק נקודת ה-send_email ב-policy שלנו. הסוכן "עובד לבד" לאורך כל ה-pipeline, אבל עוצר באדם בנקודה הקריטית — וזה מה שהופך אותו מ"אוטונומי מסוכן" ל"אוטונומי בטוח".

יש וריאציה חשובה להחלטה: לפעמים "אישור" הוא לא בינארי אלא טווח. require_approval_below: 100 אומר "קריאות מתחת ל-100 דולר מאושרות אוטומטית". require_approval_for: ["customer_id_in: [vip, enterprise]"] אומר "רק ללקוחות VIP צריך אישור". ה-policy הוא נתונים, ולכן קל להוסיף לו מורכבות בלי לשנות קוד.

4.4 למה gate דטרמיניסטי, ולא "LLM שופט בטיחות"

מפתה לחשוב: "אני אתן ל-LLM שני להחליט אם הפעולה בטוחה." זו טעות מסוכנת, משתי סיבות. ראשית, LLM שני הוא עוד משטח התקפה — אותו prompt injection ששכנע את הסוכן הראשון יכול לשכנע גם את ה"שופט". gate דטרמיניסטי, לעומת זאת, לא ניתן ל-jailbreak: rm -rf תואם את ה-regex תמיד, ללא קשר לכמה משכנע ה-context. שנית, בטיחות דורשת ודאות. "המודל החליט שזה כנראה בטוח" הוא לא תקן שאתה רוצה לשליחת email ללקוח או למחיקת DB. אתה רוצה: "הכלל המפורש אומר block/approve/allow" — חד-משמעי, ניתן ל-audit, ניתן לבדיקה ב-unit test.

זה לא אומר ש-LLM אין לו מקום בבטיחות — אבל מקומו הוא לפני ה-gate (לשפר את ההצעה) או אחרי ה-deny (להבין למה נחסם ולנסות מסלול אחר), לא בתור ה-gate עצמו. ההחלטה הסופית "לרוץ או לא לרוץ" חייבת להיות קוד. זה בדיוק העיקרון ש-Faramesh מקדם: execution control plane שאוכף gates בלי LLM שני בנתיב ההחלטה.

4.5 בדיקת policies — כי policy שלא נבדק לא עובד

policy.yaml הוא קוד — גם אם הוא נכתב ב-YAML. וקוד בלי בדיקות הוא קוד שבור שמחכה לקרות. כלל פשוט: לכל כלל ב-policy.yaml צריכה להיות בדיקה שמוודאת שהוא עושה בדיוק את מה שתיעדת. הנה מערך unit tests בסיסי:

tests/test_policy.py

import pytest
from governance.policy import PolicyGate

@pytest.fixture
def gate():
    return PolicyGate("governance/policy.yaml")

def test_blocks_destructive_bash(gate):
    d = gate.check("bash", {"command": "rm -rf ./logs"})
    assert d.action == "deny"
    assert "destructive" in d.reason

def test_blocks_destructive_sql(gate):
    d = gate.check("run_sql", {"query": "DROP TABLE users"})
    assert d.action == "deny"

def test_blocks_force_push(gate):
    d = gate.check("git", {"args": "push origin main --force"})
    assert d.action == "deny"

def test_requires_approval_for_email(gate):
    d = gate.check("send_email", {"to": "alice@example.com"})
    assert d.action == "approve"

def test_allows_safe_reads(gate):
    d = gate.check("read_file", {"path": "./workspace/notes.md"})
    assert d.action == "allow"

def test_blocks_sneaky_rm(gate):
    # variants שאנשים מנסים להתחמק
    d = gate.check("bash", {"command": "rm  -rf /tmp/x"})  # רווחים כפולים
    assert d.action == "deny", f"rm -rf variant should be blocked, got {d}"

הרץ עם pytest tests/. אם משהו נכשל — יש לך policy שלא עובד. הוסף את הריצה הזו ל-CI שלך, ולפני כל שינוי ב-policy.yaml תדע מיד אם שברת משהו.

[gate] tool=bash input={'command': 'rm -rf ./logs'} → DENY (destructive shell command)
[model received] BLOCKED by policy: destructive shell command
[model retry] tool=bash input={'command': 'ls ./logs && rm ./logs/old.log'}
[gate] tool=bash input={'command': 'ls ./logs && rm ./logs/old.log'} → ALLOW
[gate] tool=send_email input={'to': 'alice@example.com', 'subject': '...'} → APPROVE
⚠️  approval needed for send_email: outbound email needs human sign-off
    allow? [y/N] n
[model received] DENIED by human reviewer

5. parallel tool calls ו-caching — להוריד latency ועלות

שכבת ה-tools שלך עכשיו אמינה ומאובטחת. אבל היא אולי איטית. הטעות השלישית שהמחקר מסמן היא התעלמות מ-tool-call latency: קריאות tool עצמאיות שרצות בטור מייצרות תעבורת רשת עצומה וגוררות latency חמור. שתי תרופות פשוטות.

5.1 קריאות עצמאיות רצות במקביל

מודל מודרני יכול לבקש כמה tool_use blocks בתגובה אחת. אם הם עצמאיים זה מזה (לדוגמה "בדוק מלאי ב-EU" ו-"בדוק מלאי ב-US" — אף אחד לא תלוי בתוצאה של השני), אין סיבה להריץ אותם בטור. דפוס asyncio.gather (ה-מקבילה של Promise.all) מריץ את כולם יחד:

harness/parallel.py

import asyncio

async def run_parallel_tools(tool_use_blocks, run_one):
    """מריץ קריאות tool עצמאיות במקביל; מחזיר את כולן כרשימה אחת."""
    results = await asyncio.gather(*[
        run_one(block) for block in tool_use_blocks
    ])
    # חשוב: כל ה-tool_result חוזרים בהודעת user אחת
    return {"role": "user", "content": list(results)}

# שימוש בתוך הלולאה:
tool_blocks = [b for b in response.content if b.type == "tool_use"]
if tool_blocks:
    user_turn = await run_parallel_tools(tool_blocks, execute_tool_call_async)
    messages.append({"role": "assistant", "content": response.content})
    messages.append(user_turn)   # הודעה אחת, כל התוצאות

המספרים: קריאה סדרתית ל-5 tools שלוקחים 200ms כל אחד = 1 שנייה. אותם 5 במקביל = 200ms. פקטור של 5x ב-latency. בעלות — זה לא חינם, כי כל tool call עדיין צורך זמן מודל (לעבד את ה-result), אבל התקורה הזו קטנה משמעותית מזמן הרשת.

5.2 caching לפי input hash — לא לקרוא לאותו endpoint פעמיים

אם הסוכן קורא ל-get_inventory("TS-1024", "EU") פעמיים באותה ריצה, הקריאה השנייה היא בזבוז טהור — אותו input, אותו output. cache פשוט לפי hash של ה-input חוסך את הקריאה החוזרת:

import hashlib, json

class ToolCache:
    def __init__(self):
        self._store = {}

    def key(self, tool_name, tool_input):
        # מיון מפתחות מבטיח hash יציב לאותו input בכל סדר
        blob = tool_name + json.dumps(tool_input, sort_keys=True)
        return hashlib.sha256(blob.encode()).hexdigest()

    async def get_or_run(self, tool_name, tool_input, run_one):
        k = self.key(tool_name, tool_input)
        if k in self._store:
            return self._store[k]          # cache hit — אפס latency, אפס עלות
        result = await run_one(tool_name, tool_input)
        self._store[k] = result
        return result

5.3 dependency graph — מתי לא לעשות parallel

לא כל ה-tool_use blocks הם עצמאיים. לפעמים הקריאה השנייה תלויה בתוצאה של הראשונה: "חפש את המשתמש, ואז שלח לו הודעה." במקרה כזה, asyncio.gather יקרוס או יחזיר תוצאה לא נכונה. הכלל: רק קריאות עצמאיות לחלוטין (אין לאחת צורך בפלט של השנייה) רצות במקביל. קריאות תלויות חייבות לרוץ בסדר, ואת הסדר קובע המודל — הוא מבקש את הראשונה, רואה את התוצאה, ורק אז מבקש את השנייה.

מבחן מהיר: אם הייתי מריץ את הקריאה השנייה לפני הראשונה, האם הייתי מקבל תוצאה תקינה? אם כן — הן עצמאיות, parallel. אם לא — sequential. בקריאות API טיפוסיות (multi-region, multi-warehouse, multi-endpoint) התשובה היא "כן" כמעט תמיד; בקריאות שדורשות state (multi-step DB transactions) היא "לא".

☑️ צ'ק-ליסט סיום פרק

• הגדרתי custom tool עם strict: true, additionalProperties: false ו-required — ואני קורא את block.input כאובייקט, לא כמחרוזת
• מימשתי structured output על תשובת המודל עם output_config.format או messages.parse() + Pydantic
• עטפתי קריאה ב-call_with_repair עם max_repairs מפורש, וראיתי self-healing קורה ביומן (כשל בניסיון 1, תיקון בניסיון 2)
• אני יודע להבחין מתי retry מתאים (schema violation) ומתי לא (איכות / החלטה עסקית)
• הודעת השגיאה שאני מחזיר למודל כוללת את שגיאת ה-validator הגולמית + את ה-schema, לא סתם "Invalid"
• לתקלות רשת transient — הוספתי exponential backoff עם jitter, ושמרתי את ה-repair המיידי ל-schema בלבד
• חיברתי MCP filesystem server לוקלי דרך anthropic[mcp] + tool_runner והרצתי עליו קריאת tool מהלולאה
• אני יודע מתי לבחור MCP connector (server-side) ומתי client-side helpers
• וידאתי שאני לא מחזיר tool result כבד (50KB+) ישר ל-context — אלא reference ל-store חיצוני
• כתבתי policy.yaml declarative עם כללי deny לפקודות הרסניות (rm -rf, DROP TABLE) ו-require_approval לפעולות רגישות
• חיברתי את ה-PolicyGate ללולאה — לפני כל ביצוע — והוא מחזיר allow / deny / approve
• בדיקה: הסוכן ניסה פקודה הרסנית, ה-gate חסם אותו, והוא קיבל tool_result עם is_error: True במקום קריסה
• מימשתי human-approval gate על send_email (או פעולה רגישה דומה) — ה-pattern של lead-enrichment
• כתבתי unit tests (tests/test_policy.py) שמכסים כל כלל ב-policy.yaml, והם רצים ב-pytest
• אני מריץ קריאות tool עצמאיות במקביל ב-asyncio.gather, ומחזיר את כל ה-tool_result בהודעת user אחת
• הוספתי cache לפי input hash ל-tools idempotent בלבד — ולא ל-tools עם side-effect
• עברתי על שגרת העבודה (schema → output → retry → gate → tests → latency) ואני בונה tools חדשים בסדר הזה