Wrap a deny.sh vault entry as a CrewAI tool.

# pip install crewai deny-sh requests
import os, requests, json
from crewai import Agent, Task, Crew
from crewai.tools import tool
from deny_sh import vault_get

@tool("Get Stripe Invoice")
def get_invoice(id: str) -> str:
    """Look up a Stripe invoice by id and return its amount_due and status."""
    # 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 to the agent
    return json.dumps({
        "id": body.get("id"),
        "amount_due": body.get("amount_due"),
        "status": body.get("status"),
    })

billing_agent = Agent(
    role="Billing Assistant",
    goal="Look up invoice details accurately.",
    backstory="You are a billing assistant with read-only access to Stripe invoices.",
    tools=[get_invoice],
)
lookup_task = Task(
    description="Look up invoice in_1Abc123XYZ and summarise it.",
    expected_output="A concise summary of the invoice amount and status.",
    agent=billing_agent,
)
crew = Crew(agents=[billing_agent], tasks=[lookup_task])
result = crew.kickoff()
print(result)

# pip install crewai deny-sh requests pydantic
import os, requests, json
from typing import Type
from pydantic import BaseModel, Field
from crewai import Agent, Task, Crew
from crewai.tools import BaseTool
from deny_sh import vault_get

class InvoiceInput(BaseModel):
    id: str = Field(description="The Stripe invoice ID to look up")

class GetInvoiceTool(BaseTool):
    name: str = "get_invoice"
    description: str = (
        "Look up a Stripe invoice by id and return its amount_due and status."
    )
    args_schema: Type[BaseModel] = InvoiceInput

    def _run(self, id: str) -> str:
        # 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: only the fields the agent needs
        return json.dumps({
            "id": body.get("id"),
            "amount_due": body.get("amount_due"),
            "status": body.get("status"),
        })

billing_agent = Agent(
    role="Billing Assistant",
    goal="Look up invoice details accurately.",
    backstory="You are a billing assistant with read-only access to Stripe invoices.",
    tools=[GetInvoiceTool()],
)
lookup_task = Task(
    description="Look up invoice in_1Abc123XYZ and summarise it.",
    expected_output="A concise summary of the invoice amount and status.",
    agent=billing_agent,
)
crew = Crew(agents=[billing_agent], tasks=[lookup_task])
result = crew.kickoff()
print(result)

def tools_for_tenant(tenant_pw: str, tenant_api_key: str):
    @tool("Get Stripe Invoice")
    def get_invoice(id: str) -> str:
        """Look up a Stripe invoice by id."""
        # tenant_pw and tenant_api_key captured in closure — no shared key
        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 [get_invoice]

# per-request: build agent + crew fresh with this tenant's tools
agent = Agent(role="Billing Assistant", goal="Look up invoice details.",
    backstory="Expert billing agent.",
    tools=tools_for_tenant(tenant_pw, tenant_api_key))
task = Task(description="Look up invoice in_1Abc123XYZ.",
    expected_output="Invoice amount and status.", agent=agent)
result = Crew(agents=[agent], tasks=[task]).kickoff()