Guarded Run

Once you have a GuardedRun from guard.run(), use model_call() and tool_call() to wrap your model and tool invocations. Each call is checked against your guardrails before execution.

Model Calls

async with guard.run(user_id="alice") as run:
    response = await run.model_call(
        fn=lambda: openai.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": "Hello"}],
        ),
        model="gpt-4o",
        input_data={"messages": [{"role": "user", "content": "Hello"}]},
        token_extractor=lambda r: (
            r.usage.prompt_tokens,
            r.usage.completion_tokens,
        ),
    )

Parameters

Parameter	Type	Required	Description
`fn`	`Callable[[], T \| Awaitable[T]]`	Yes	The function that makes the model call. Can be sync or async.
`model`	`str`	Yes	Model identifier (e.g., `"gpt-4o"`, `"claude-sonnet-4-5-20250929"`). Used for cost calculation.
`input_data`	`dict[str, Any] \| None`	No	Input payload to record in the step (for audit).
`token_extractor`	`Callable[[T], tuple[int, int]] \| None`	No	Extracts `(prompt_tokens, completion_tokens)` from the response.

What happens

Check: POST /v1/runs/{id}/steps with type MODEL_CALL — guardrails are evaluated
Execute: fn() is called (awaited if async)
Report: PATCH /v1/runs/{id}/steps/{step_id} with status, duration, and token counts

If the guardrail check denies the step, PikarcBlockedError is raised before fn() executes — your model call never happens, and you’re never billed by the provider.

Token Extractor

The token_extractor function receives the return value of fn() and should return a tuple of (prompt_tokens, completion_tokens). This is how Pikarc tracks token usage and calculates costs. For OpenAI:

token_extractor=lambda r: (r.usage.prompt_tokens, r.usage.completion_tokens)

For Anthropic:

token_extractor=lambda r: (r.usage.input_tokens, r.usage.output_tokens)

If omitted, token counts won’t be recorded for the step.

Tool Calls

async with guard.run(user_id="alice") as run:
    result = await run.tool_call(
        fn=lambda: search_database(query="recent orders"),
        tool_name="search_database",
        input_data={"query": "recent orders"},
    )

Parameters

Parameter	Type	Required	Description
`fn`	`Callable[[], T \| Awaitable[T]]`	Yes	The function that executes the tool. Can be sync or async.
`tool_name`	`str`	Yes	Tool identifier (e.g., `"search_database"`, `"send_email"`).
`input_data`	`dict[str, Any] \| None`	No	Input payload to record in the step (for audit).

What happens

Check: POST /v1/runs/{id}/steps with type TOOL_CALL — guardrails are evaluated
Execute: fn() is called (awaited if async)
Report: PATCH /v1/runs/{id}/steps/{step_id} with status and duration

Tool calls don’t have a token_extractor since tools don’t produce token counts.

Sync and Async Functions

Both model_call() and tool_call() accept sync or async callables for fn. The SDK detects whether the return value is awaitable and handles it automatically:

# Sync function — works fine
await run.tool_call(
    fn=lambda: my_sync_tool(arg),
    tool_name="my_sync_tool",
)

# Async function — also works
await run.tool_call(
    fn=lambda: my_async_tool(arg),
    tool_name="my_async_tool",
)

Full Example

from pikarc import AsyncPikarc, PikarcBlockedError
from openai import AsyncOpenAI

openai = AsyncOpenAI()
guard = AsyncPikarc(api_key="lg_...", base_url="http://localhost:8000")

try:
    async with guard.run(user_id="alice", metadata={"agent": "support-bot"}) as run:
        # Model call — guardrail-checked, tokens tracked
        response = await run.model_call(
            fn=lambda: openai.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": "Find recent orders"}],
            ),
            model="gpt-4o",
            token_extractor=lambda r: (
                r.usage.prompt_tokens,
                r.usage.completion_tokens,
            ),
        )

        # Tool call — guardrail-checked, duration tracked
        orders = await run.tool_call(
            fn=lambda: search_orders(user="alice", limit=10),
            tool_name="search_orders",
            input_data={"user": "alice", "limit": 10},
        )

        # Another model call — guardrails checked again
        summary = await run.model_call(
            fn=lambda: openai.chat.completions.create(
                model="gpt-4o",
                messages=[
                    {"role": "user", "content": "Find recent orders"},
                    {"role": "assistant", "content": response.choices[0].message.content},
                    {"role": "user", "content": f"Summarize: {orders}"},
                ],
            ),
            model="gpt-4o",
            token_extractor=lambda r: (
                r.usage.prompt_tokens,
                r.usage.completion_tokens,
            ),
        )

except PikarcBlockedError as e:
    print(f"Blocked: {e.reason} ({e.deny_reason})")

finally:
    await guard.close()

Getting Started

SDK

API Reference

Guides

Configuration

Model Calls

Parameters

What happens

Token Extractor

Tool Calls

Parameters

What happens

Sync and Async Functions

Full Example

​Model Calls

​Parameters

​What happens

​Token Extractor

​Tool Calls

​Parameters

​What happens

​Sync and Async Functions

​Full Example

Model Calls

Parameters

What happens

Token Extractor

Tool Calls

Parameters

What happens

Sync and Async Functions

Full Example