""" A governed CrewAI crew — a two-agent travel booking, end to end. The developer pages name OpenAI Agents SDK and LangGraph with runnable examples beside them. CrewAI was named in the same breath, by the same persona research, and had nothing anyone could run — a search for "CrewAI" returned MCP and nothing else. Naming a framework in prose is not an integration. CrewAI's own pitch is multiple agents cooperating on one goal, so this crew is two agents, not one: a researcher that finds flights (read-only) and a booker that purchases one (consequential). Only the second needs a gate — the shape below is unchanged from the other two examples for that reason: ┌────────────┐ ┌──────┐ ┌──────────────────┐ ┌─────────────┐ │ booker agent│ ───▶ │ tool │ ───▶ │ guarded callable │ ───▶ │ your airline│ └────────────┘ └──────┘ └────────┬─────────┘ └─────────────┘ │ asks the gate first; │ a refusal raises and the ▼ real function is never called MetaMynd authorize Governance is not a step in the crew's process, and it is not a Task `guardrail` — a guardrail can inspect what a task OUTPUT (text); it cannot see the signed action, the mandate, or the amount. It wraps the tool itself, so the only path to the real function goes through the gate, same as the LangGraph and OpenAI Agents SDK examples. Requires: crewai, cryptography. pip install crewai cryptography export METAMYND_API=https://metamynd.ai/api/v1 export AGENT_DID=did:hedera:testnet:... export AGENT_KEY=302e020100... # as issued python crewai_agent.py Provision an agent in about four seconds, no account: npx create-metamynd-agent --sandbox An LLM key is needed only to actually run the crew (`crew.kickoff()`, not called by this file's demo — see the note at the end). The governance runs without one. Verified against crewai 1.15.18 and the live gate. """ from __future__ import annotations import os from typing import Any from metamynd_client import GovernanceBlocked, MetaMyndClient, guard_tool # --------------------------------------------------------------------------------------- # 1. Your tools, exactly as you already have them. Nothing governance-aware here. # --------------------------------------------------------------------------------------- def search_flights(route: str) -> list[dict[str, Any]]: """Search available flights for a route. Read-only, and deliberately ungoverned — see the note at the bottom.""" return [ {"airline": "skyward-air", "route": route, "amount": 420.0}, {"airline": "northwind-rail", "route": route, "amount": 180.0}, ] def book_flight(airline: str, amount: float, route: str) -> dict[str, Any]: """Book a flight and charge the corporate card. Replace the body with your real call.""" return {"pnr": "QK7T2M", "airline": airline, "amount": amount, "route": route} # --------------------------------------------------------------------------------------- # 2. Wrap the consequential one, THEN decorate. This is the whole integration. # # CrewAI's tool() decorator builds the tool's argument schema from inspect.signature(), # the same mechanism the OpenAI Agents SDK example warns about — a wrapper that copies # only __name__ and __doc__ would present itself as (*args, **kwargs) with no fields for # the agent to fill in. guard_tool uses functools.wraps precisely so this cannot happen: # verified directly against this crewai version before writing this file — # # governed = guard_tool(client, "flight-purchase", book_flight, ...) # tool(governed).args_schema.model_fields.keys() # -> {'airline', 'amount', 'route'} # # identical to tool(book_flight) itself # # So: guard on the INSIDE, because the gate must be inside the callable CrewAI invokes; # the decorator on the OUTSIDE, because it describes the tool to the agent. # --------------------------------------------------------------------------------------- # Built only when the credentials are present, so importing this file never fails: `main()` says what is # missing instead of dying at import with a stack trace. client = MetaMyndClient.from_env() if os.getenv("AGENT_DID") and os.getenv("AGENT_KEY") else None # type: ignore[assignment] governed_book_flight = guard_tool( client, "flight-purchase", # = your mandate scope book_flight, # Map the tool's arguments onto what the gate decides against. Get this wrong and the # gate authorizes an amount of zero against an empty merchant, which passes a cap and # an allow-list both — so the mapping is part of the control, not plumbing around it. lambda airline, amount, route: { "amount": amount, "merchant": airline, # `riskLevel` is REQUIRED by the starter rules: a call that carries none is escalated, not allowed (a call # that hides its risk is indistinguishable from one lying about it). 'low' here is a PLACEHOLDER, not an # assessment - a real integration states an honest one, or lets the mandate's owner set a `riskTier` # the agent cannot lower. "context": {"tool": "flight-purchase", "route": route, "riskLevel": "low"}, }, ) def build_tools() -> dict[str, Any]: """The CrewAI tool objects. Imported lazily so this file runs without crewai installed.""" from crewai.tools import tool return { "search": tool(search_flights), "book": tool(governed_book_flight), } def build_crew() -> Any: """An ordinary two-agent CrewAI crew. The governance is already done.""" from crewai import Agent, Crew, Process, Task tools = build_tools() researcher = Agent( role="Flight researcher", goal="Find available flights for the requested route", backstory="You search airline inventory and report options, without booking anything.", tools=[tools["search"]], ) booker = Agent( role="Flight booker", goal="Book the requested flight within policy", backstory=( "You book flights for the team. If a booking is refused, explain why to the " "requester rather than trying another way — do not retry with different arguments." ), tools=[tools["book"]], ) research_task = Task( description="Find flight options for {route}.", expected_output="A short list of airlines, routes and prices.", agent=researcher, ) booking_task = Task( description="Book the best flight found for {route}, charging {airline} for {amount}.", expected_output="A confirmation PNR, or a clear explanation of why booking was refused.", agent=booker, context=[research_task], ) return Crew(agents=[researcher, booker], tasks=[research_task, booking_task], process=Process.sequential) # --------------------------------------------------------------------------------------- # 3. What it does when you run it # # Three refusal shapes, because "it blocks things" is not the interesting claim. Each of # these fails for a structurally different reason, and only the second is a spend limit. # The booker agent's TOOL is called directly here rather than through crew.kickoff() — see # the note at the end for why, and what changes when a real LLM chooses the arguments. # --------------------------------------------------------------------------------------- def attempt(label: str, call: Any) -> None: """Run one case and report the outcome without letting a refusal end the program.""" print(f" {label}") try: result = call() print(f" -> booked {result['pnr']}\n") except GovernanceBlocked as e: print(f" -> {e.verdict.decision}/{e.verdict.reason_code} — book_flight was never called\n") def main() -> None: if not os.getenv("AGENT_DID") or not os.getenv("AGENT_KEY"): raise SystemExit("Set AGENT_DID and AGENT_KEY — run `npx create-metamynd-agent --sandbox` for a free pair.") print("\n A governed CrewAI crew\n") # Inside the mandate: named airline, under the cap. The tool runs. attempt( "1. $420 on skyward-air — within the mandate", lambda: governed_book_flight(airline="skyward-air", amount=420, route="KUL-SIN"), ) # Over the per-transaction cap. The airline call never happens. attempt( "2. $2,400 on skyward-air — over the per-transaction cap", lambda: governed_book_flight(airline="skyward-air", amount=2_400, route="KUL-LHR"), ) # Under the cap, but the merchant was never named. A cap alone would have allowed this. attempt( "3. $95 on a merchant the mandate never listed", lambda: governed_book_flight(airline="unlisted-charter", amount=95, route="KUL-PEN"), ) # The one a prompt could not have stopped. attempt( "4. the booker agent asks to raise its own limit", lambda: guard_tool(client, "permissions.update", lambda **_: {"pnr": "n/a"})(limit=100_000), ) print(" (nothing forbids that action; the mandate simply never granted it)\n") # And the crew, to show the wiring is unchanged. try: crew = build_crew() booker_tools = next(a for a in crew.agents if a.role == "Flight booker").tools params = list(booker_tools[0].args_schema.model_fields.keys()) print(f" the booker agent's tool takes: book_flight({', '.join(params)})") print(" identical to the unguarded function, which is the point\n") except ImportError: print(" (pip install crewai to build the crew; the guarding above is the integration)\n") # --------------------------------------------------------------------------------------- # A note on why this calls the tool directly instead of crew.kickoff() # # A real crew needs an LLM key and network time neither of which this file should require # to demonstrate governance — the researcher and booker agents above are real, buildable # CrewAI agents, and crew.kickoff({"route": "KUL-SIN"}) runs them exactly as shown once you # have a key configured. What changes when you do: the BOOKER AGENT, not this file, chooses # the airline/amount/route arguments, from the researcher's findings and its own # instructions. The gate call inside governed_book_flight does not change at all — it does # not know or care whether a human, a script, or an LLM produced the arguments it is asked # to authorize. That is the property this example exists to demonstrate: governance sits # at the tool boundary, not in the crew's reasoning, so it holds regardless of who or what # is doing the reasoning. # # A note on Task guardrails, because you will reach for them # # CrewAI's Task accepts a `guardrail` — a function that validates a task's OUTPUT before # it is accepted, and can ask the agent to retry. Genuinely useful, and not this. A # guardrail inspects what the booker agent SAID it did (or the text it produced); it has # no visibility into the signed action, the mandate that authorized it, or the amount # actually charged. The gate decides an ACTION, signed by the agent's key against a # mandate its owner issued, before the airline is ever called — cases 3 and 4 above were # refused with no rule written against an unlisted charter or a self-granted limit, and # no task description mentioned either, because the mandate simply never granted them. # A guardrail could not have caught either: nothing in the booker's output looks wrong. # # A note on what is NOT governed here # # search_flights is unwrapped on purpose. Governing every call would add a network round # trip to a read that cannot spend money or change anything, and a control that makes the # ordinary path slow is a control teams remove. Wrap what is consequential. # # The corollary is worth stating plainly, because nothing will warn you: an unwrapped tool # is an ungoverned tool. The gate can only refuse calls that reach it. Deciding which of # your tools are consequential is a design step, not a configuration one — and # `npx @metamynd/agentsafe-guard verify` will tell you what your mandate actually # constrains, including the controls it does not set. # --------------------------------------------------------------------------------------- if __name__ == "__main__": main()