CrewAI Integration

CrewAI’s multi-agent framework lets you build teams of agents working together on complex tasks. The managed-agents package governs every tool call any crew member makes — before it executes, with a full audit trail per agent.

Installation

pip install managed-agents

Governing a CrewAI tool

Wrap your CrewAI tools with governance before they execute. Each crew member should have its own harness with its own agent ID so you can track each member’s behavior separately in the audit trail:

python

from managed_agents import NomoticHarness, GovernanceDenied
from crewai_tools import BaseTool

researcher_harness = NomoticHarness(
    api_key="nm_live_...",
    agent_id="nmc-researcher-...",
    platform="crewai",
)

class GovernedSearchTool(BaseTool):
    name: str = "web_search"
    description: str = "Search the web for information"

    async def _arun(self, query: str) -> str:
        await researcher_harness.govern(
            "web_search",
            {"query": query}
        )
        return await self.perform_search(query)

    def _run(self, query: str) -> str:
        import asyncio
        asyncio.run(
            researcher_harness.govern("web_search", {"query": query})
        )
        return self.perform_search(query)

Separate harnesses per crew member

Giving each crew member its own Nomotic agent ID means each has its own certificate, trust score, and behavioral history in your dashboard. This makes it clear which crew member took which action — essential for audit trail integrity.

python

researcher_harness = NomoticHarness(
    api_key="nm_live_...",
    agent_id="nmc-researcher-...",
    platform="crewai",
)

writer_harness = NomoticHarness(
    api_key="nm_live_...",
    agent_id="nmc-writer-...",
    platform="crewai",
)

Handling denials

When GovernanceDenied is raised, return a safe fallback string rather than letting the exception propagate to CrewAI’s orchestration layer:

python

async def _arun(self, query: str) -> str:
    try:
        await researcher_harness.govern("web_search", {"query": query})
        return await self.perform_search(query)
    except GovernanceDenied as e:
        return f"Search not permitted by governance policy. Reason: {e}"