CrewAI Integration

pip install agentsecretstore crewai crewai-tools

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from agentsecretstore import AgentVault
import asyncio

class VaultToolInput(BaseModel):
    path: str = Field(description="The vault secret path (e.g. 'production/gemini/GEMINI_API_KEY')")

class AgentSecretStoreTool(BaseTool):
    name: str = "vault_get_secret"
    description: str = (
        "Retrieve a secret or API key from Agent Secret Store secure vault. "
        "Always use this before calling any external API that requires authentication. "
        "Never ask users for credentials — fetch them from the vault."
    )
    args_schema: type[BaseModel] = VaultToolInput

    agent_key: str  # Injected by orchestrator

    def _run(self, path: str) -> str:
        """Fetch secret from vault synchronously."""

        async def fetch():
            async with AgentVault(agent_key=self.agent_key) as vault:
                return await vault.get_secret(path)

        loop = asyncio.new_event_loop()
        try:
            return loop.run_until_complete(fetch())
        finally:
            loop.close()

import asyncio
import os
from crewai import Agent, Task, Crew, Process, LLM
from agentsecretstore import AgentVault

# ── Orchestrator: provide least-privilege agent keys per role ─

def provision_agent_keys() -> dict[str, str]:
    """Load one least-privilege Agent Secret Store key per crew member."""

    return {
        "research":  os.environ["ASS_RESEARCH_AGENT_KEY"],
        "writer":    os.environ["ASS_WRITER_AGENT_KEY"],
        "publisher": os.environ["ASS_PUBLISHER_AGENT_KEY"],
    }

# ── Vault-aware tools ─────────────────────────────────────────

def get_gemini_key(agent_key: str) -> str:
    """Fetch Gemini key from the vault."""
    async def fetch():
        async with AgentVault(agent_key=agent_key) as vault:
            return await vault.get_secret("production/gemini/GEMINI_API_KEY")
    return asyncio.run(fetch())

# ── Build the crew ────────────────────────────────────────────

async def run_crew():
    agent_keys = provision_agent_keys()

    async with AgentVault(agent_key=agent_keys["writer"]) as vault:
        gemini_key = await vault.get_secret("production/gemini/GEMINI_API_KEY")
    gemini_llm = LLM(model="google/gemini-2.0-flash", api_key=gemini_key)

    vault_tools = {
        role: AgentSecretStoreTool(agent_key=agent_key)
        for role, agent_key in agent_keys.items()
    }

    # ── Agents ───────────────────────────────────────────────
    researcher = Agent(
        role="Research Analyst",
        goal="Find comprehensive, accurate information on the given topic",
        backstory="An expert researcher who always uses proper data sources.",
        tools=[vault_tools["research"]],
        llm=gemini_llm,
        verbose=True,
    )

    writer = Agent(
        role="Content Writer",
        goal="Transform research into compelling, well-structured content",
        backstory="A skilled technical writer who produces clear, accurate content.",
        tools=[vault_tools["writer"]],
        llm=gemini_llm,
        verbose=True,
    )

    publisher = Agent(
        role="Content Publisher",
        goal="Distribute finished content to the appropriate channels",
        backstory="A distribution specialist who ensures content reaches the right audience.",
        tools=[vault_tools["publisher"]],
        llm=gemini_llm,
        verbose=True,
    )

    # ── Tasks ─────────────────────────────────────────────────
    research_task = Task(
        description="Research the latest developments in AI agent security",
        expected_output="A detailed report with key findings and source citations",
        agent=researcher,
    )

    writing_task = Task(
        description="Write a technical blog post based on the research report",
        expected_output="A 1500-word blog post in Markdown format",
        agent=writer,
        context=[research_task],
    )

    publish_task = Task(
        description="Publish the blog post to Slack #engineering channel",
        expected_output="Confirmation that the post was published successfully",
        agent=publisher,
        context=[writing_task],
    )

    # ── Run the crew ─────────────────────────────────────────
    crew = Crew(
        agents=[researcher, writer, publisher],
        tasks=[research_task, writing_task, publish_task],
        process=Process.sequential,
        verbose=True,
    )

    result = crew.kickoff()
    return result

if __name__ == "__main__":
    asyncio.run(run_crew())

# Pattern: parallel crew with shared vault tool
# Each task runs concurrently — each worker has its own agent key

import os
from crewai import Agent, Task, Crew, Process

async def run_parallel_research_crew():
    agent_keys = [
        os.environ["ASS_RESEARCH_WORKER_1_KEY"],
        os.environ["ASS_RESEARCH_WORKER_2_KEY"],
        os.environ["ASS_RESEARCH_WORKER_3_KEY"],
    ]

    agents = [
        Agent(
            role=f"Research Worker {i+1}",
            goal=f"Research topic segment {i+1}",
            tools=[AgentSecretStoreTool(agent_key=agent_key)],
        )
        for i, agent_key in enumerate(agent_keys)
    ]

    # Tasks can run in parallel
    tasks = [
        Task(
            description=f"Research topic: AI agent frameworks (segment {i+1})",
            expected_output="Detailed findings for this segment",
            agent=agent,
        )
        for i, agent in enumerate(agents)
    ]

    crew = Crew(
        agents=agents,
        tasks=tasks,
        process=Process.parallel,  # Run research tasks concurrently
    )

    return crew.kickoff()