Pydantic AI Integration

from stackone_ai import StackOneToolSet
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel

def create_agent_for_account(account_id: str):
    """
    Create agent with tools for a specific account.

    In production, account_id comes from:
    - User/tenant context
    - Authentication middleware
    - Database lookup
    """
    # Initialize toolset
    toolset = StackOneToolSet()

    # Fetch tools dynamically for this account
    tools = toolset.fetch_tools(account_ids=[account_id])

    # Create Pydantic AI agent
    agent = Agent(
        model=OpenAIModel('gpt-5'),
        system_prompt="You are a helpful HR assistant with access to employee data."
    )

    # Register StackOne tools
    openai_tools = tools.to_openai()
    for tool_def in openai_tools:
        tool_name = tool_def['function']['name']
        tool = tools.get_tool(tool_name)

        # Create callable function
        def make_tool_func(t=tool):
            def tool_func(**kwargs):
                result = t.call(**kwargs)
                return result.data if hasattr(result, 'data') else result
            return tool_func

        agent.tool(make_tool_func(), name=tool_name)

    return agent

# Use the agent
account_id = get_current_user_account()  # Your function
agent = create_agent_for_account(account_id)

result = agent.run_sync("List employees in engineering department")
print(result.data)

from stackone_ai import StackOneToolSet

toolset = StackOneToolSet()

# Load all tools for specific accounts (most common)
tools = toolset.fetch_tools(
    account_ids=["account-123", "account-456"]
)

# Optional: Filter by operation type for security
tools = toolset.fetch_tools(
    account_ids=["account-123"],
    actions=["*_list_*", "*_get_*"]  # Only list and get operations (read-only)
)

from pydantic import BaseModel
from pydantic_ai import Agent

class EmployeeSummary(BaseModel):
    total_count: int
    departments: list[str]
    average_tenure_years: float

agent = Agent(
    model=OpenAIModel('gpt-5'),
    result_type=EmployeeSummary,
    system_prompt="Analyze employee data and return structured summaries."
)

# Returns validated EmployeeSummary instance
result = agent.run_sync("Summarize the employee base")
summary: EmployeeSummary = result.data

print(f"Total employees: {summary.total_count}")
print(f"Departments: {', '.join(summary.departments)}")

# Get from user/tenant context
account_id = request.user.stackone_account_id
account_id = get_account_for_tenant(tenant_id)

from stackone_ai.exceptions import StackOneError, ToolExecutionError

try:
    result = agent.run_sync(message)
except ToolExecutionError as e:
    print(f"Tool execution failed: {e.message}")
except StackOneError as e:
    print(f"StackOne API error: {e.message}")

# Only expose read operations (optional - for security)
tools = toolset.fetch_tools(
    account_ids=[account_id],
    actions=["*_list_*", "*_get_*"]  # Only list and get
)

# Or exclude dangerous operations
tools = toolset.fetch_tools(
    account_ids=[account_id],
    actions=["!*_delete_*"]  # Everything except deletes
)