CrewAI¶
mimir.integrations.crewai.as_crewai_tool converts a decision tool into a BaseTool whose result_schema is the typed result, for use in a CrewAI Agent's tool list.
tool_call_hook(check, approve=...) returns a hook for register_before_tool_call_hook: a certified denial blocks the call, and an escalated call is forwarded to your approve function, which receives the tool name and arguments and returns whether the call may run.
Tested from crewai 1.15.
Note. CrewAI pins the MCP SDK to version 1, which conflicts with
mimir-decisions[mcp]. Install CrewAI separately from the MCP extra. The MCP example for CrewAI connects over HTTP rather than stdio for this reason.
Native tools¶
"""A CrewAI crew that routes support tickets with a MIMIR decision tool.
Install `mimir-decisions[local,crewai]`, set `OPENAI_API_KEY`, then run
`make example NAME=crewai/agent`.
"""
from typing import Final
from crewai import LLM, Agent, Crew, Task
from crewai.llms.base_llm import BaseLLM
from mimir import Choice, Decider, Mimir
from mimir.integrations.crewai import as_crewai_tool
MODEL: Final = "openai/gpt-5.5"
TEAMS: Final = {
"billing": "Billing: payments, refunds and invoices",
"security": "Security: account access, passwords and fraud",
"shipping": "Shipping: deliveries, tracking and returns",
}
def build_crew(decider: Decider, llm: BaseLLM | None = None) -> Crew:
"""A one-agent crew routing the ticket given as the `ticket` input, answering with `decider`."""
route_ticket = decider.tool(
"route_ticket",
Choice("Which team should handle this ticket?", TEAMS),
description="Route a customer support ticket to the team that owns it.",
)
agent = Agent(
role="Support dispatcher",
goal="Send every ticket to the team that owns it.",
backstory="You route tickets with route_ticket and escalate deferred decisions.",
llm=LLM(model=MODEL) if llm is None else llm,
tools=[as_crewai_tool(route_ticket)],
)
task = Task(
description="Route this ticket: {ticket}",
expected_output="The team that will answer, or that a person will review the ticket.",
agent=agent,
)
return Crew(agents=[agent], tasks=[task])
def main() -> None:
crew = build_crew(Mimir.from_pretrained("Mythologic/MIMIR-1"))
print(crew.kickoff(inputs={"ticket": "I was charged twice for order 4412."}))
if __name__ == "__main__":
main()
Tool-call check¶
"""A CrewAI crew whose refunds MIMIR checks against the refund rules first.
A certified yes runs the refund, a certified no blocks it, and anything else asks a person at
the console. The check is a global before-tool-call hook, registered while the crew runs.
Install `mimir-decisions[local,crewai]`, set `OPENAI_API_KEY`, then run
`make example NAME=crewai/guarded_agent`.
"""
from typing import Final
from crewai import LLM, Agent, Crew, Task
from crewai.hooks import (
ToolCallHookContext,
register_before_tool_call_hook,
unregister_before_tool_call_hook,
)
from crewai.llms.base_llm import BaseLLM
from crewai.tools import BaseTool, tool
from mimir import CheckOutcome, Decider, Mimir
from mimir.integrations.crewai import tool_call_hook
MODEL: Final = "openai/gpt-5.5"
RULES: Final = (
"A refund is at most the amount the customer paid for the order.",
"Refunds above 500 dollars need a manager's approval.",
)
@tool("issue_refund")
def issue_refund(order: str, amount: float) -> str:
"""Refund `amount` dollars on `order`."""
return f"Refunded {amount:.2f} dollars on order {order}."
def ask_at_the_console(context: ToolCallHookContext, outcome: CheckOutcome) -> bool:
"""Whether the person at the console approves the escalated call."""
answer = context.request_human_input(
prompt=f"{outcome.reason} Approve {context.tool_name} {context.tool_input}?",
default_message="Type 'y' to approve:",
)
return answer.strip().lower() == "y"
def build_crew(refund: BaseTool = issue_refund, llm: BaseLLM | None = None) -> Crew:
"""A one-agent crew issuing the refund given as the `request` input."""
agent = Agent(
role="Refunds agent",
goal="Issue the refunds customers ask for.",
backstory="You issue refunds with issue_refund.",
llm=LLM(model=MODEL) if llm is None else llm,
tools=[refund],
)
task = Task(description="{request}", expected_output="What was refunded.", agent=agent)
return Crew(agents=[agent], tasks=[task])
def run_checked(crew: Crew, decider: Decider, request: str) -> str:
"""Run `crew` with its refunds checked by `decider` against `RULES`."""
check = decider.tool_call_check(RULES, tools=["issue_refund"])
hook = tool_call_hook(check, approve=ask_at_the_console)
register_before_tool_call_hook(hook)
try:
return str(crew.kickoff(inputs={"request": request}))
finally:
unregister_before_tool_call_hook(hook)
def main() -> None:
decider = Mimir.from_pretrained("Mythologic/MIMIR-1")
print(run_checked(build_crew(), decider, "Refund 900 dollars on order 4412."))
if __name__ == "__main__":
main()
Over MCP¶
"""A CrewAI crew using a running MIMIR MCP server over Streamable HTTP, through `mcps=`.
CrewAI names MCP tools after their server: over HTTP the agent sees
`127_0_0_1_8000_mcp_route_ticket`, while over stdio the whole launch command goes into the
name and truncates it past recognition. Start the server with
`make mcp TOOLS=examples/tools.yaml`, set `OPENAI_API_KEY`, then run
`make example NAME=crewai/mcp_agent`. With `MIMIR_API_KEYS` set on the server, pass one of the
keys as `api_key`.
"""
from typing import Final
from crewai import LLM, Agent, Crew, Task
from crewai.llms.base_llm import BaseLLM
from crewai.mcp import MCPServerHTTP
MODEL: Final = "openai/gpt-5.5"
URL: Final = "http://127.0.0.1:8000/mcp"
def mimir_server(url: str = URL, api_key: str | None = None) -> MCPServerHTTP:
"""The MIMIR MCP server at `url`, authorised with `api_key` when the server has keys."""
headers = None if api_key is None else {"Authorization": f"Bearer {api_key}"}
return MCPServerHTTP(url=url, headers=headers)
def build_crew(server: MCPServerHTTP, llm: BaseLLM | None = None) -> Crew:
"""A one-agent crew routing the ticket given as the `ticket` input, with the server's tools."""
agent = Agent(
role="Support dispatcher",
goal="Send every ticket to the team that owns it.",
backstory="You route tickets with the route_ticket tool and escalate deferred decisions.",
llm=LLM(model=MODEL) if llm is None else llm,
mcps=[server],
)
task = Task(
description="Route this ticket: {ticket}",
expected_output="The team that will answer, or that a person will review the ticket.",
agent=agent,
)
return Crew(agents=[agent], tasks=[task])
def main() -> None:
crew = build_crew(mimir_server())
print(crew.kickoff(inputs={"ticket": "I was charged twice for order 4412."}))
if __name__ == "__main__":
main()