Pydantic AI Integration

import json
from stackone_ai import StackOneToolSet
from pydantic_ai import Agent, Tool

def create_agent_for_account(account_id: str):
    # Initialize toolset and fetch tools
    toolset = StackOneToolSet()
    tools = toolset.fetch_tools(account_ids=[account_id])

    # Convert each StackOne tool to a Pydantic AI Tool with proper schema
    pydantic_tools = []
    for stackone_tool in tools:
        params_schema = stackone_tool.to_openai_function()["function"]["parameters"]

        def execute(t=stackone_tool, **kwargs: object) -> str:
            return json.dumps(t.execute(kwargs))

        pydantic_tools.append(
            Tool.from_schema(
                execute,
                name=stackone_tool.name,
                description=stackone_tool.description,
                json_schema=params_schema,
            )
        )

    # Create agent with tools
    agent = Agent(
        "openai:gpt-5.4",
        system_prompt="You are a helpful HR assistant.",
        tools=pydantic_tools,
    )
    return agent

# Get account ID from your app's auth context or StackOne dashboard
account_id = "your-account-id"
agent = create_agent_for_account(account_id)

result = agent.run_sync("List the first 5 employees")
print(result.output)

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.4'),
    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.models import StackOneError, StackOneAPIError

try:
    result = agent.run_sync(message)
except StackOneAPIError as e:
    print(f"API error: {e.message}")
except StackOneError as e:
    print(f"StackOne 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
)