Per-agent credential isolation for CrewAI multi-agent systems
pip install crewai-avpIn CrewAI systems, multiple agents collaborate on tasks. Each agent may need different credentials with different permission levels:
from crewai import Agent, Task, Crew
from crewai_avp import AVPAgentCredentials
# Create credential manager
creds = AVPAgentCredentials(
vault_path="~/.avp/vault.enc",
password="your-password"
)
# Create agents with isolated credentials
researcher = Agent(
role="Research Analyst",
goal="Find relevant information",
backstory="Expert at finding data",
llm=creds.get_llm("researcher", provider="openai"),
tools=[creds.get_tool("researcher", "serp_search")]
)
writer = Agent(
role="Content Writer",
goal="Write compelling content",
backstory="Expert writer",
llm=creds.get_llm("writer", provider="anthropic")
)
# Create crew
crew = Crew(
agents=[researcher, writer],
tasks=[...]
)
result = crew.kickoff()Each agent gets its own isolated workspace. The researcher agent cannot access the writer's credentials and vice versa.
Get pre-configured LLM instances with credentials automatically injected from the vault.
Tools receive their required API keys automatically based on the agent's workspace.
Track which agent accessed which credential and when.
The main class for managing agent credentials.
from crewai_avp import AVPAgentCredentials, Backend
# File backend (default)
creds = AVPAgentCredentials(
vault_path="~/.avp/vault.enc",
password="your-password"
)
# Keychain backend
creds = AVPAgentCredentials(backend=Backend.KEYCHAIN)
# Memory backend (testing)
creds = AVPAgentCredentials(backend=Backend.MEMORY)# Store credentials for an agent workspace
creds.store("researcher", "SERP_API_KEY", "...")
creds.store("researcher", "ARXIV_TOKEN", "...")
creds.store("writer", "OPENAI_API_KEY", "sk-...")
# Get an LLM configured with agent's credentials
llm = creds.get_llm("researcher", provider="openai", model="gpt-4")
# Get a tool configured with agent's credentials
search_tool = creds.get_tool("researcher", "serp_search")
web_tool = creds.get_tool("researcher", "web_scraper")
# Get raw credential value (use sparingly)
api_key = creds.get("researcher", "SERP_API_KEY")
# List credentials for an agent
keys = creds.list("researcher")
# ["SERP_API_KEY", "ARXIV_TOKEN"]
# Check if credential exists
if creds.has("researcher", "SERP_API_KEY"):
print("Key is configured")# OpenAI
llm = creds.get_llm("agent", provider="openai", model="gpt-4")
# Looks for: OPENAI_API_KEY
# Anthropic
llm = creds.get_llm("agent", provider="anthropic", model="claude-3-opus")
# Looks for: ANTHROPIC_API_KEY
# Azure OpenAI
llm = creds.get_llm("agent", provider="azure", model="gpt-4")
# Looks for: AZURE_OPENAI_API_KEY, AZURE_OPENAI_ENDPOINT
# Google
llm = creds.get_llm("agent", provider="google", model="gemini-pro")
# Looks for: GOOGLE_API_KEY
# Ollama (local, no credentials needed)
llm = creds.get_llm("agent", provider="ollama", model="llama2")# Search tools
serp = creds.get_tool("agent", "serp_search") # SERP_API_KEY
ddg = creds.get_tool("agent", "duckduckgo") # No key needed
brave = creds.get_tool("agent", "brave_search") # BRAVE_API_KEY
# Web tools
scraper = creds.get_tool("agent", "web_scraper") # No key needed
browser = creds.get_tool("agent", "browser") # BROWSERLESS_API_KEY
# Database tools
sql = creds.get_tool("agent", "sql", db_uri="...")
mongodb = creds.get_tool("agent", "mongodb", connection_string="...")
# API tools
github = creds.get_tool("agent", "github") # GITHUB_TOKEN
jira = creds.get_tool("agent", "jira") # JIRA_TOKEN, JIRA_URL
slack = creds.get_tool("agent", "slack") # SLACK_BOT_TOKEN"""CrewAI research team with AVP credential management."""
from crewai import Agent, Task, Crew, Process
from crewai_avp import AVPAgentCredentials
# Initialize credential manager
creds = AVPAgentCredentials(
vault_path="~/.avp/research-crew.enc",
password="secure-password"
)
# Store credentials (do this once during setup)
# creds.store("researcher", "SERP_API_KEY", "...")
# creds.store("researcher", "OPENAI_API_KEY", "sk-...")
# creds.store("analyst", "OPENAI_API_KEY", "sk-...")
# creds.store("writer", "ANTHROPIC_API_KEY", "sk-ant-...")
# Create agents with isolated credentials
researcher = Agent(
role="Senior Research Analyst",
goal="Uncover cutting-edge developments in AI",
backstory="""You work at a leading tech think tank.
Your expertise lies in identifying emerging trends.""",
verbose=True,
allow_delegation=False,
llm=creds.get_llm("researcher", provider="openai", model="gpt-4"),
tools=[
creds.get_tool("researcher", "serp_search"),
creds.get_tool("researcher", "web_scraper")
]
)
analyst = Agent(
role="Data Analyst",
goal="Analyze research data and extract insights",
backstory="""You are an expert data analyst with a keen eye
for patterns and trends in complex datasets.""",
verbose=True,
allow_delegation=False,
llm=creds.get_llm("analyst", provider="openai", model="gpt-4")
)
writer = Agent(
role="Tech Content Strategist",
goal="Craft compelling content on tech advancements",
backstory="""You are a renowned Content Strategist, known for
your insightful and engaging articles.""",
verbose=True,
allow_delegation=True,
llm=creds.get_llm("writer", provider="anthropic", model="claude-3-opus")
)
# Define tasks
research_task = Task(
description="""Conduct comprehensive research on the latest
AI agent frameworks in 2024. Focus on key players, features,
and market trends.""",
expected_output="A detailed research report with citations",
agent=researcher
)
analysis_task = Task(
description="""Analyze the research findings and identify
the top 3 emerging trends in AI agent development.""",
expected_output="Analysis report with trend predictions",
agent=analyst
)
writing_task = Task(
description="""Using the research and analysis, write an
engaging blog post about the future of AI agents.""",
expected_output="A 1500-word blog post in markdown format",
agent=writer
)
# Create and run crew
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
process=Process.sequential,
verbose=2
)
result = crew.kickoff()
print(result)
# Clean up
creds.close()Create a setup script to initialize credentials for your crew:
#!/usr/bin/env python3
"""Setup script for CrewAI credentials."""
from crewai_avp import AVPAgentCredentials
import getpass
def setup_credentials():
password = getpass.getpass("Vault password: ")
creds = AVPAgentCredentials(
vault_path="~/.avp/research-crew.enc",
password=password
)
# Researcher credentials
creds.store("researcher", "OPENAI_API_KEY",
getpass.getpass("Researcher OpenAI key: "))
creds.store("researcher", "SERP_API_KEY",
getpass.getpass("Researcher SERP key: "))
# Analyst credentials
creds.store("analyst", "OPENAI_API_KEY",
getpass.getpass("Analyst OpenAI key: "))
# Writer credentials
creds.store("writer", "ANTHROPIC_API_KEY",
getpass.getpass("Writer Anthropic key: "))
print("Credentials stored successfully!")
creds.close()
if __name__ == "__main__":
setup_credentials()