Resolve vault credentials inside an AutoGen tool function.

# pip install autogen-agentchat autogen-ext[openai] deny-sh requests
import asyncio, os, requests, json
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from deny_sh import vault_get

async def get_invoice(id: str) -> str:
    """Look up a Stripe invoice by id and return status and amount due."""
    # resolve the credential INSIDE the tool boundary
    stripe_key = vault_get("stripe-prod", os.environ["VAULT_PW"])
    r = requests.get(
        f"https://api.stripe.com/v1/invoices/{id}",
        headers={"Authorization": f"Bearer {stripe_key}"},
    )
    body = r.json()
    # narrowed DTO: never return raw upstream response bodies
    return json.dumps({
        "id": body.get("id"),
        "amount_due": body.get("amount_due"),
        "status": body.get("status"),
    })

model_client = OpenAIChatCompletionClient(model="gpt-4o")

agent = AssistantAgent(
    name="billing_agent",
    model_client=model_client,
    tools=[get_invoice],
    system_message="You are a billing assistant. Use tools to look up invoices.",
)

async def main():
    result = await agent.run(task="Look up invoice in_1Abc123XYZ")
    print(result.messages[-1].content)

asyncio.run(main())

from autogen_agentchat.base import TaskResult
from autogen_agentchat.messages import AgentEvent, ChatMessage

async def main_stream():
    async for event in agent.run_stream(task="Look up invoice in_1Abc123XYZ"):
        if isinstance(event, (AgentEvent, ChatMessage)):
            print(f"[{event.source}] {event.content}", flush=True)
        elif isinstance(event, TaskResult):
            print(f"\nFinal: {event.messages[-1].content}")
            break

asyncio.run(main_stream())

def build_agent_for_tenant(tenant_pw: str, tenant_api_key: str) -> AssistantAgent:
    async def get_invoice(id: str) -> str:
        """Look up a Stripe invoice for this tenant by id."""
        # tenant_pw and tenant_api_key captured in closure — never shared
        stripe_key = vault_get(
            "stripe-prod",
            tenant_pw,
            api_key=tenant_api_key,
        )
        r = requests.get(
            f"https://api.stripe.com/v1/invoices/{id}",
            headers={"Authorization": f"Bearer {stripe_key}"},
        )
        body = r.json()
        return json.dumps({
            "id": body.get("id"),
            "amount_due": body.get("amount_due"),
            "status": body.get("status"),
        })

    return AssistantAgent(
        name="billing_agent",
        model_client=OpenAIChatCompletionClient(model="gpt-4o"),
        tools=[get_invoice],
        system_message="You are a billing assistant.",
    )

# per-request: resolve tenant credentials from your own store, then build
agent = build_agent_for_tenant(tenant_pw=pw, tenant_api_key=key)
result = await agent.run(task="Look up invoice in_1Abc123XYZ")