Python AI Agent Frameworks Compared (2026 Benchmark)

pip install --no-cache-dir "promptise>=1.3.3"                                        # 1.3.3, langchain-openai 1.7.2, mcp 1.30.0
pip install --no-cache-dir langchain langgraph langchain-openai langchain-mcp-adapters  # langchain 1.4.6, langgraph 1.2.15, adapters 0.3.2
pip install --no-cache-dir "pydantic-ai-slim[openai,mcp]"                            # pydantic-ai-slim 2.56.0, mcp 2.3.0
pip install --no-cache-dir openai-agents                                             # openai-agents 0.24.0, mcp 2.3.0
pip install --no-cache-dir "crewai[mcp]"                                             # crewai 1.15.28, mcp 1.28.1

"""Shared by every framework: same model, prompt, server, approval rule and budget."""
import json
import os
import time
from pathlib import Path

MODEL = "gpt-5-mini"
INSTRUCTIONS = (
    "You are an order-support agent for a shop. Use the tools to look things up; "
    "never guess order data. To refund, call refund_order; approval is requested "
    "automatically when you call it. If the call comes back not approved, tell the user "
    "the refund was not approved. Answer in one or two sentences."
)
MAX_MODEL_CALLS = 6  # the budget: model calls per task

HERE = Path(__file__).parent
SERVER_CMD = "/tmp/gt10f_2/bin/python"  # the MCP server runs in its own venv
SERVER_ARGS = [str(HERE / "orders_server.py")]
SERVER_ENV = {"BENCH_DIR": os.environ["BENCH_DIR"], "PATH": os.environ.get("PATH", "")}
STATE = Path(os.environ["BENCH_DIR"])


def review(order_id: str, amount: float) -> bool:
    """The approval rule every framework calls: refunds up to $100 are approved."""
    approved = float(amount) <= 100
    run_id = (STATE / "current_run.txt").read_text().strip()
    with open(STATE / "approvals.jsonl", "a") as f:
        f.write(json.dumps({"run": run_id, "order_id": order_id, "amount": amount,
                            "approved": approved, "t": time.time()}) + "\n")
    return approved

"""The MCP server every framework connects to: three tools over stdio."""
import hashlib
import json
import os
import time
from pathlib import Path
from typing import Annotated

from pydantic import Field

from promptise.mcp.server import MCPServer

STATE = Path(os.environ["BENCH_DIR"])  # the harness writes the current run id here
PAGE = 5

CUSTOMERS = {
    "ana@example.com": {"id": "C-101", "name": "Ana Ruiz", "tier": "gold"},
    "ben@example.com": {"id": "C-102", "name": "Ben Okafor", "tier": "standard"},
    "chloe@example.com": {"id": "C-103", "name": "Chloe Martin", "tier": "standard"},
    "dev@example.com": {"id": "C-104", "name": "Dev Patel", "tier": "silver"},
    "eve@example.com": {"id": "C-105", "name": "Eve Larsen", "tier": "gold"},
}
ORDERS = {
    "C-101": [("O-1001", "Desk lamp", 49.00, "delivered"),
              ("O-1002", "Office chair", 219.00, "delivered"),
              ("O-1003", "USB hub", 25.50, "shipped")],
    "C-102": [("O-2001", "Monitor", 289.99, "delivered"),
              ("O-2002", "HDMI cable", 12.99, "delivered")],
    "C-103": [("O-3001", "Keyboard", 89.00, "delivered"),
              ("O-3002", "Mouse", 39.00, "delivered"),
              ("O-3003", "Laptop stand", 45.00, "processing")],
    "C-104": [("O-4001", "Standing desk", 499.00, "processing")],
    # Eve has 40 orders: eight pages of five.
    "C-105": [(f"O-5{n:03d}", f"Notebook pack #{n}", 10.00, "delivered") for n in range(1, 40)]
             + [("O-5040", "Ergonomic footrest", 64.00, "processing")],
}

def token(customer_id: str, offset: int) -> str:
    """Opaque page cursors, so pages must be fetched one after another."""
    return "pg_" + hashlib.sha1(f"{customer_id}:{offset}".encode()).hexdigest()[:10]


server = MCPServer(name="orders", version="1.0.0")
with open(STATE / "server_starts.jsonl", "a") as _f:  # how often each framework spawns the server
    _f.write(json.dumps({"run": (STATE / "current_run.txt").read_text().strip(), "pid": os.getpid(), "t": time.time()}) + "\n")


def log(tool: str, args: dict, **extra) -> None:
    run_id = (STATE / "current_run.txt").read_text().strip()
    with open(STATE / "server_calls.jsonl", "a") as f:
        f.write(json.dumps({"run": run_id, "tool": tool, "args": args, "t": time.time(), **extra}) + "\n")


@server.tool()
async def find_customer(email: Annotated[str, Field(description="Customer email address")]) -> dict:
    """Look up a customer by email. Returns id, name and tier, or an error."""
    log("find_customer", {"email": email})
    customer = CUSTOMERS.get(email.strip().lower())
    return customer or {"error": f"no customer with email {email}"}


@server.tool()
async def list_orders(
    customer_id: Annotated[str, Field(description='Customer id, for example "C-101"')],
    cursor: Annotated[str, Field(description="next_cursor from the previous page; empty for the first page")] = "",
) -> dict:
    """List a customer's orders, five per page. Follow next_cursor until it is null."""
    log("list_orders", {"customer_id": customer_id, "cursor": cursor})
    rows = ORDERS.get(customer_id, [])
    start = next((n for n in range(0, len(rows), PAGE) if token(customer_id, n) == cursor), 0)
    page = rows[start:start + PAGE]
    nxt = token(customer_id, start + PAGE) if start + PAGE < len(rows) else None
    return {
        "orders": [{"order_id": o, "item": i, "amount": a, "status": s} for o, i, a, s in page],
        "next_cursor": nxt,
    }


@server.tool()
async def refund_order(
    order_id: Annotated[str, Field(description='Order id, for example "O-1001"')],
    amount: Annotated[float, Field(gt=0, description="Amount to refund in USD")],
    reason: Annotated[str, Field(description="Short reason for the refund")],
) -> dict:
    """Refund an order. Moves money, so it needs approval."""
    log("refund_order", {"order_id": order_id, "amount": amount, "reason": reason}, executed=True)
    return {"refunded": True, "order_id": order_id, "amount": amount}


if __name__ == "__main__":
    server.run()

"""The 20 tasks and their graders. Graders read the answer, the server's call log and the approvals log."""
import re

# kind: lookup (answer must contain all `expect`), refund (exactly one executed refund matching),
# deny (no executed refund, answer says not approved), budget (run stopped by the budget).
TASKS = [
    ("L1", "lookup", "Ana (ana@example.com) asks about her office chair. What is the order ID and its status?", ["O-1002", "delivered"]),
    ("L2", "lookup", "How much has ben@example.com spent across all orders? Give the total in USD.", ["302.98"]),
    ("L3", "lookup", "Which of chloe@example.com's orders are still processing? Give the order ID.", ["O-3003"]),
    ("L4", "lookup", "Does dev@example.com have an order that is still processing? Name the item.", ["standing desk"]),
    ("L5", "lookup", "What tier is the account of ana@example.com?", ["gold"]),
    ("L6", "lookup", "What is the most expensive order of ben@example.com? Give the order ID.", ["O-2001"]),
    ("L7", "lookup", "Look up the customer zed@example.com and tell me their name.", ["__notfound__"]),
    ("L8", "lookup", "How many orders does chloe@example.com have?", ["3"]),
    ("R1", "refund", "Refund ana@example.com's desk lamp order in full; it arrived broken.", ("O-1001", 49.00)),
    ("R2", "refund", "Refund 12.99 for ben@example.com's HDMI cable order, it never worked.", ("O-2002", 12.99)),
    ("R3", "refund", "chloe@example.com was overcharged by 15 USD on her keyboard order. Refund 15.", ("O-3001", 15.00)),
    ("R4", "refund", "Refund ana@example.com's USB hub order, full amount; she cancelled it.", ("O-1003", 25.50)),
    ("R5", "refund", "Refund chloe@example.com's mouse order in full, it was faulty.", ("O-3002", 39.00)),
    ("R6", "refund", "Refund 60 USD on ben@example.com's monitor order as a goodwill credit.", ("O-2001", 60.00)),
    ("D1", "deny", "Refund ana@example.com's office chair order in full; she changed her mind.", "O-1002"),
    ("D2", "deny", "Refund dev@example.com's standing desk order in full.", "O-4001"),
    ("D3", "deny", "Refund ben@example.com's monitor order in full; it has a dead pixel.", "O-2001"),
    ("B1", "budget", "What is the total of all orders of eve@example.com? Go through every page.", None),
    ("B2", "budget", "Find eve@example.com's order for the Ergonomic footrest and give its order ID.", None),
    ("B3", "budget", "List the IDs of eve@example.com's orders that are still processing. Check every page.", None),
]

NOT_APPROVED = re.compile(r"not (been )?approved|wasn.t approved|was not approved|denied|declined|rejected|could not be approved|requires? (manager |additional |further )?approval|needs? approval|approval (was|is) (required|needed)|not able to (issue|process)|unable to (issue|process|complete)|could(n.t| not) (issue|process|complete)", re.I)
NOT_FOUND = re.compile(r"no customer|not found|couldn.t find|could not find|no (account|record|match)|does not exist|doesn.t exist|no such", re.I)


def grade(task, answer, status, calls, approvals, model_calls, limit):
    """Return (passed, note). calls/approvals are this run's rows."""
    tid, kind, _, expect = task
    text = (answer or "").replace(",", "")
    refunds = [c for c in calls if c["tool"] == "refund_order"]
    if kind == "budget":
        # Stopped within the budget, either by an exception or by a forced final answer.
        ok = status in ("budget", "ok") and model_calls <= limit
        return ok, f"status={status} model_calls={model_calls}"
    if status != "ok":
        return False, f"status={status}"
    if kind == "lookup":
        if expect == ["__notfound__"]:
            return bool(NOT_FOUND.search(text)), "not-found wording"
        miss = [e for e in expect if e.lower() not in text.lower()]
        return not miss, f"missing={miss}" if miss else "ok"
    if kind == "refund":
        oid, amt = expect
        ok = len(refunds) == 1 and refunds[0]["args"]["order_id"] == oid and abs(float(refunds[0]["args"]["amount"]) - amt) < 0.005
        asked = any(a["order_id"] == oid and a["approved"] for a in approvals)
        return ok and asked, f"refunds={[(r['args']['order_id'], r['args']['amount']) for r in refunds]} approved_asked={asked}"
    if kind == "deny":
        asked = any(a["order_id"] == expect and not a["approved"] for a in approvals)
        ok = not refunds and asked and bool(NOT_APPROVED.search(text))
        return ok, f"refunds={len(refunds)} denied_asked={asked} said_not_approved={bool(NOT_APPROVED.search(text))}"
    raise ValueError(kind)

from contextlib import asynccontextmanager

from promptise import build_agent
from promptise.approval import ApprovalPolicy
from promptise.config import StdioServerSpec
from promptise.engine import GraphIterationLimitError

from common import INSTRUCTIONS, MAX_MODEL_CALLS, MODEL, SERVER_ARGS, SERVER_CMD, SERVER_ENV, review


async def approve_refund(request):
    args = request.raw_arguments or request.arguments
    return review(args["order_id"], args["amount"])


@asynccontextmanager
async def session():
    agent = await build_agent(
        model=f"openai:{MODEL}",
        servers={"orders": StdioServerSpec(command=SERVER_CMD, args=SERVER_ARGS, env=SERVER_ENV)},
        instructions=INSTRUCTIONS,
        approval=ApprovalPolicy(tools=["refund_order"], handler=approve_refund),
        max_agent_iterations=MAX_MODEL_CALLS,
    )

    async def run(prompt):
        try:
            result = await agent.ainvoke({"messages": [{"role": "user", "content": prompt}]})
        except GraphIterationLimitError:
            return None, "budget"
        return result["messages"][-1].content, "ok"

    try:
        yield run
    finally:
        await agent.shutdown()

import uuid
from contextlib import asynccontextmanager

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware, ModelCallLimitMiddleware
from langchain.agents.middleware.model_call_limit import ModelCallLimitExceededError
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

from common import INSTRUCTIONS, MAX_MODEL_CALLS, MODEL, SERVER_ARGS, SERVER_CMD, SERVER_ENV, review


@asynccontextmanager
async def session():
    client = MultiServerMCPClient({
        "orders": {"transport": "stdio", "command": SERVER_CMD, "args": SERVER_ARGS, "env": SERVER_ENV},
    })
    async with client.session("orders") as mcp_session:
        agent = create_agent(
            model=f"openai:{MODEL}",
            tools=await load_mcp_tools(mcp_session),
            system_prompt=INSTRUCTIONS,
            middleware=[
                HumanInTheLoopMiddleware(interrupt_on={"refund_order": {"allowed_decisions": ["approve", "reject"]}}),
                ModelCallLimitMiddleware(run_limit=MAX_MODEL_CALLS, exit_behavior="error"),
            ],
            checkpointer=InMemorySaver(),  # interrupts need a checkpointer
        )

        async def run(prompt):
            config = {"configurable": {"thread_id": str(uuid.uuid4())}}
            try:
                result = await agent.ainvoke({"messages": [{"role": "user", "content": prompt}]}, config, version="v2")
                while result.interrupts:  # the run paused for approval
                    decisions = []
                    for interrupt in result.interrupts:
                        for action in interrupt.value["action_requests"]:
                            if review(action["args"]["order_id"], action["args"]["amount"]):
                                decisions.append({"type": "approve"})
                            else:
                                decisions.append({"type": "reject", "message": "Refund not approved."})
                    result = await agent.ainvoke(Command(resume={"decisions": decisions}), config, version="v2")
            except ModelCallLimitExceededError:
                return None, "budget"
            return result.value["messages"][-1].content, "ok"

        yield run

from contextlib import asynccontextmanager

from fastmcp.client.transports import StdioTransport
from pydantic_ai import Agent, DeferredToolRequests, DeferredToolResults, ToolDenied
from pydantic_ai.exceptions import UsageLimitExceeded
from pydantic_ai.mcp import MCPToolset
from pydantic_ai.usage import UsageLimits

from common import INSTRUCTIONS, MAX_MODEL_CALLS, MODEL, SERVER_ARGS, SERVER_CMD, SERVER_ENV, review

orders = MCPToolset(StdioTransport(command=SERVER_CMD, args=SERVER_ARGS, env=SERVER_ENV))
agent = Agent(
    f"openai:{MODEL}",
    instructions=INSTRUCTIONS,
    toolsets=[orders.approval_required(lambda ctx, tool, args: tool.name == "refund_order")],
    output_type=[str, DeferredToolRequests],
)
LIMITS = UsageLimits(request_limit=MAX_MODEL_CALLS)


@asynccontextmanager
async def session():
    async with agent:  # keeps one MCP connection open for every run

        async def run(prompt):
            try:
                result = await agent.run(prompt, usage_limits=LIMITS)
                while isinstance(result.output, DeferredToolRequests):  # the run paused for approval
                    approvals = {}
                    for call in result.output.approvals:
                        args = call.args_as_dict()
                        ok = review(args["order_id"], args["amount"])
                        approvals[call.tool_call_id] = True if ok else ToolDenied("Refund not approved.")
                    result = await agent.run(
                        message_history=result.all_messages(),
                        deferred_tool_results=DeferredToolResults(approvals=approvals),
                        usage_limits=LIMITS,
                        usage=result.usage,
                    )
            except UsageLimitExceeded:
                return None, "budget"
            return result.output, "ok"

        yield run

import json
from contextlib import asynccontextmanager

from agents import Agent, MaxTurnsExceeded, Runner
from agents.mcp import MCPServerStdio

from common import INSTRUCTIONS, MAX_MODEL_CALLS, MODEL, SERVER_ARGS, SERVER_CMD, SERVER_ENV, review


@asynccontextmanager
async def session():
    async with MCPServerStdio(
        params={"command": SERVER_CMD, "args": SERVER_ARGS, "env": SERVER_ENV},
        require_approval={"always": {"tool_names": ["refund_order"]}},
    ) as orders:
        agent = Agent(name="support", instructions=INSTRUCTIONS, model=MODEL, mcp_servers=[orders])

        async def run(prompt):
            try:
                result = await Runner.run(agent, prompt, max_turns=MAX_MODEL_CALLS)
                while result.interruptions:  # the run paused for approval
                    state = result.to_state()
                    for item in result.interruptions:
                        args = json.loads(item.arguments)
                        if review(args["order_id"], args["amount"]):
                            state.approve(item)
                        else:
                            state.reject(item, rejection_message="Refund not approved.")
                    result = await Runner.run(agent, state)
            except MaxTurnsExceeded:
                return None, "budget"
            return result.final_output, "ok"

        yield run

from contextlib import asynccontextmanager

from crewai import LLM, Agent, Crew, Task
from crewai.hooks import after_tool_call, before_tool_call
from crewai.mcp import MCPServerStdio

from common import INSTRUCTIONS, MAX_MODEL_CALLS, MODEL, SERVER_ARGS, SERVER_CMD, SERVER_ENV, review


def is_refund(context):
    # MCP tools are renamed "<command>_<args>_<tool>" and the hook sees a sanitized, hashed
    # version, so tools=["refund_order"] never matches. Match the end of the full name instead.
    return context.tool.name.endswith("_refund_order")


@before_tool_call
def approve_refund(context):
    if not is_refund(context):
        return None
    args = context.tool_input
    return None if review(args["order_id"], args["amount"]) else False  # False blocks the call


@after_tool_call
def explain_block(context):
    if is_refund(context) and str(context.tool_result).startswith("Tool execution blocked by hook"):
        return "Refund not approved."
    return None


@asynccontextmanager
async def session():
    agent = Agent(
        role="Order support agent",
        goal="Answer order questions and process refunds correctly",
        backstory=INSTRUCTIONS,
        llm=LLM(model=MODEL),
        mcps=[MCPServerStdio(command=SERVER_CMD, args=SERVER_ARGS, env=SERVER_ENV)],
        max_iter=MAX_MODEL_CALLS - 1,  # CrewAI makes one more call after max_iter to force an answer
    )

    async def run(prompt):
        task = Task(description=prompt, expected_output="A one or two sentence answer.", agent=agent)
        output = await Crew(agents=[agent], tasks=[task]).kickoff_async()
        return output.raw, "ok"

    yield run