AutoGen Integration

pip install agentsecretstore pyautogen

import asyncio
from typing import Annotated
import autogen
from agentsecretstore import AgentVault

def make_vault_tool(vault_token: str):
    """
    Factory that returns a vault retrieval function bound to a scoped token.
    The token — not the master key — is what the agent uses.
    """

    async def get_secret(
        path: Annotated[str, "Secret path in the vault, e.g. 'production/openai/api-key'"],
    ) -> str:
        """
        Retrieve a credential from Agent Secret Store.
        Call this before any operation that requires an API key or credential.
        Returns the secret value as a plain string.
        """
        async with AgentVault() as vault:
            secret = await vault.get_secret(path, token=vault_token)
        return secret.value

    return get_secret

import asyncio
import autogen
from agentsecretstore import AgentVault

async def run_research_agent():
    """AutoGen agent that fetches all credentials from the vault at runtime."""

    # Step 1: Orchestrator issues a scoped token for this agent run
    async with AgentVault() as vault:
        token = await vault.request_token(
            scope="secrets:read:production/*",
            ttl_seconds=1800,
            description="AutoGen research agent session",
        )

        # Also fetch the LLM key directly for the assistant config
        openai_key = await vault.get_secret("production/openai/api-key")

    # Step 2: Configure the LLM using the vault-fetched key
    llm_config = {
        "config_list": [
            {
                "model": "gpt-4o",
                "api_key": openai_key.value,  # from vault, not env var
            }
        ],
        "temperature": 0.1,
        "timeout": 120,
    }

    # Step 3: Create the vault retrieval function bound to the scoped token
    get_secret = make_vault_tool(vault_token=token.value)

    # Step 4: Register the vault tool with AutoGen
    user_proxy = autogen.UserProxyAgent(
        name="user_proxy",
        human_input_mode="NEVER",
        max_consecutive_auto_reply=10,
        is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
        code_execution_config={"use_docker": False},
    )

    assistant = autogen.AssistantAgent(
        name="research_assistant",
        llm_config=llm_config,
        system_message=(
            "You are an AI research assistant with access to a secure credential vault. "
            "When you need to call any external API, always use the get_secret tool first "
            "to retrieve the required credential. Never ask the user for API keys. "
            "Reply TERMINATE when the task is complete."
        ),
    )

    # Register the vault tool for both calling and execution
    autogen.register_function(
        get_secret,
        caller=assistant,
        executor=user_proxy,
        name="get_secret",
        description="Retrieve an API key or credential from the secure vault by path",
    )

    # Step 5: Run the agent
    await user_proxy.a_initiate_chat(
        assistant,
        message=(
            "Research the latest developments in multi-agent AI frameworks. "
            "Use the Tavily search API (key at 'production/tavily/api-key') "
            "to find recent papers and summarize the top 3 findings."
        ),
    )

asyncio.run(run_research_agent())

import asyncio
import autogen
from agentsecretstore import AgentVault

async def run_multi_agent_pipeline():
    """
    Multi-agent AutoGen pipeline where each agent has its own scoped token.
    The researcher can only read search credentials.
    The writer can only read publishing credentials.
    Neither can access the other's secrets.
    """

    async with AgentVault() as vault:
        # Fetch shared LLM key
        openai_key = await vault.get_secret("production/openai/api-key")

        # Issue separate scoped tokens per agent — least privilege
        researcher_token = await vault.request_token(
            scope="secrets:read:production/search/*",
            ttl_seconds=3600,
            description="AutoGen researcher agent",
        )
        writer_token = await vault.request_token(
            scope="secrets:read:production/publishing/*",
            ttl_seconds=3600,
            description="AutoGen writer agent",
        )
        critic_token = await vault.request_token(
            scope="secrets:read:production/quality/*",
            ttl_seconds=3600,
            description="AutoGen critic agent",
        )

    llm_config = {
        "config_list": [{"model": "gpt-4o", "api_key": openai_key.value}],
    }

    # Each agent gets its own vault tool bound to its scoped token
    researcher_vault = make_vault_tool(vault_token=researcher_token.value)
    writer_vault = make_vault_tool(vault_token=writer_token.value)
    critic_vault = make_vault_tool(vault_token=critic_token.value)

    # Researcher: can only access production/search/* secrets
    researcher = autogen.AssistantAgent(
        name="researcher",
        llm_config=llm_config,
        system_message=(
            "You research topics using web search APIs. "
            "Use get_secret to retrieve 'production/search/tavily-key' before searching."
        ),
    )

    # Writer: can only access production/publishing/* secrets
    writer = autogen.AssistantAgent(
        name="writer",
        llm_config=llm_config,
        system_message=(
            "You write and publish content. "
            "Use get_secret to retrieve 'production/publishing/wordpress-key' before posting."
        ),
    )

    # Critic: can only access quality service credentials
    critic = autogen.AssistantAgent(
        name="critic",
        llm_config=llm_config,
        system_message=(
            "You review content quality. "
            "Use get_secret for 'production/quality/grammarly-key' to check writing."
        ),
    )

    user_proxy = autogen.UserProxyAgent(
        name="user_proxy",
        human_input_mode="NEVER",
        max_consecutive_auto_reply=20,
        code_execution_config={"use_docker": False},
    )

    # Register scoped vault tools per agent — researcher can't use writer's tool
    autogen.register_function(
        researcher_vault, caller=researcher, executor=user_proxy,
        name="get_secret", description="Retrieve a search API credential",
    )
    autogen.register_function(
        writer_vault, caller=writer, executor=user_proxy,
        name="get_secret", description="Retrieve a publishing credential",
    )
    autogen.register_function(
        critic_vault, caller=critic, executor=user_proxy,
        name="get_secret", description="Retrieve a quality service credential",
    )

    # Run the group chat
    groupchat = autogen.GroupChat(
        agents=[user_proxy, researcher, writer, critic],
        messages=[],
        max_round=12,
    )
    manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config)

    await user_proxy.a_initiate_chat(
        manager,
        message="Research and write a blog post about AutoGen multi-agent patterns.",
    )

asyncio.run(run_multi_agent_pipeline())

import asyncio
from typing import Annotated
import autogen
from agentsecretstore import AgentVault

# Advanced pattern: tool that verifies scope before returning
def make_audited_vault_tool(vault_token: str, agent_name: str):
    """Enhanced vault tool that logs the requesting agent name."""

    async def get_secret(
        path: Annotated[str, "Vault secret path"],
        reason: Annotated[str, "Why this secret is needed (recorded in audit)"] = "",
    ) -> str:
        """Retrieve a credential. Provide a reason for audit clarity."""
        async with AgentVault() as vault:
            secret = await vault.get_secret(
                path,
                token=vault_token,
                # Extra metadata recorded in the audit log
                metadata={"agent_name": agent_name, "reason": reason},
            )
        return secret.value

    return get_secret