Documentation Index
Fetch the complete documentation index at: https://docs.pikarc.dev/llms.txt
Use this file to discover all available pages before exploring further.
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.
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)
token_extractor=lambda r: (r.usage.input_tokens, r.usage.output_tokens)
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()
