petfishFramework

A lightweight Python framework for reliable, auditable, budget-aware, and permission-aware AI agents.

Status: Alpha — API may change. Core runtime works; see Roadmap.

Quick Start (Zero Cost — No API Key)

pip install petfishframework

from petfishframework import Agent, ReAct
from petfishframework.tools.calculator import Calculator
from petfishframework.models.fake import FakeModel

# FakeModel — runs without any API key, perfect for testing
model = FakeModel.script_tool_then_answer(
    tool_name="calculator",
    tool_args={"expression": "17 * 23"},
    final_answer="391",
)

agent = Agent(
    model=model,
    reasoning=ReAct(),
    tools=(Calculator(),),
)

result = agent.run("What is 17 * 23?")
print(result.answer)  # "391"
print(result.usage.total_tokens)
print(len(result.trajectory.steps), "steps")

Quick Start (Real LLM)

pip install "petfishframework[openai]"

from petfishframework import Agent, ReAct
from petfishframework.tools.calculator import Calculator
from petfishframework.models.openai import OpenAIModel

agent = Agent(
    model=OpenAIModel(model="gpt-4o-mini"),  # or model="openai:gpt-4o-mini"
    reasoning=ReAct(),
    tools=(Calculator(),),
)

result = agent.run("What is 17 * 23?")
print(result.answer)  # "391"

Set OPENAI_API_KEY in .env or environment. Works with OpenAI-compatible APIs (SiliconFlow, etc.) via OPENAI_BASE_URL.

Budget Control (Runtime Hard Limits)

from petfishframework import Budget

# Budget is execution-scoped — pass to run() or session(), not Agent()
result = agent.run(
    "Complex calculation task",
    budget=Budget(max_tokens=1000, max_tool_calls=5, max_steps=10),
)
# Exceeding any limit raises BudgetExceeded

Permission Gate (Runtime Access Control)

from petfishframework.permissions.model import (
    Decision, DecisionEffect, PermissionPolicy,
)

class DenyExpensiveTools:
    """Custom policy: deny tools tagged 'expensive'."""
    def evaluate(self, subject, action, resource, context):
        if "expensive" in resource.tags:
            return Decision(effect=DecisionEffect.DENY, reason="too expensive")
        return Decision(effect=DecisionEffect.ALLOW)

agent = Agent(
    model=model,
    reasoning=ReAct(),
    tools=(Calculator(),),
    permission_policy=DenyExpensiveTools(),
)
# Tool calls pass through the Environment chokepoint — denied calls never execute

Replay & Audit (Event-Sourced Sessions)

session = agent.session("What is 17 * 23?")
result = session.run()

# Every step is recorded — model calls, tool calls, permission decisions
for event in session.replay():
    print(f"{event.type}: {event.data}")

# Events come from session.replay(), not from Result
# Result has: answer, usage, trajectory
# Session has: events, replay(), checkpoint()

MCP Client (External Tool Servers)

pip install "petfishframework[mcp]"

from petfishframework.mcp import connect_stdio

# Connect to a real MCP server
client = connect_stdio("npx", ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"])
tools = client.discover_tools()  # 14 tools: read_file, write_file, list_directory...

agent = Agent(model=model, reasoning=ReAct(), tools=tuple(tools))
result = agent.run("List all files in /tmp")

Reliability Evaluation (Pass^k)

from petfishframework.reliability import pass_at_k_with_perturbations, exact_match
from petfishframework.core.types import Task

# Run same task 8 times — measure consistency
result = pass_at_k_with_perturbations(
    session_factory=lambda task: agent.session(task),
    task=Task(prompt="What is 17 * 23?"),
    k=8,
)
print(result.summary())
# Pass@8 — PASS (100%)
#   canonical:        8/8
#   order_shuffled:   8/8
#   paraphrase:       8/8

Core Concepts

Features

• 3 reasoning strategies: ReAct, LATS (MCTS search), LLM+P (symbolic planning)
• 3 model adapters: OpenAI, Anthropic, FakeModel (deterministic testing)
• 3 routing axes: ToolRegistry (auto tool selection), Adaptive-RAG (retrieval), ReasoningStrategy
• MCP client: real stdio transport, tool discovery from external MCP servers
• Multi-agent: AgentAsTool (supervisor delegates to specialist agents)
• Structured output: JSON → dataclass (zero regex scoring)
• Conversation memory: cross-session recall via ConversationStore
• Async + streaming: dual sync/async interface

Documentation

• Usage Guide — full lifecycle, 18 sections
• API Reference — 989-line definitive reference
• Architecture — 5 core decisions
• Benchmark Results — 3-tier strategy
• Examples — quickstart, tools+retrieval, multi-agent

Enterprise PoC

See examples/05_enterprise_expense.py and tests/test_enterprise_demo.py for a complete enterprise expense approval scenario demonstrating all 6 DecisionEffects.

Run (from a repository checkout): python examples/05_enterprise_expense.py

YAML Policy Engine

Configure permissions via YAML instead of Python:

from petfishframework.policies import YamlPolicy

policy = YamlPolicy.from_file("examples/policies/enterprise-expense.yaml")
policy.register_tools(tools)

agent = Agent(model=model, reasoning=ReAct(), tools=tools, permission_policy=policy)

Supported conditions: action.tool_name, subject.role_in, subject.role_not_in, action.args.amount_gt/lt, tool.side_effect, tool.external_egress.

CredentialBroker

Tools never hold raw API keys — they receive scoped, time-limited tokens:

from petfishframework.credentials import CredentialBroker

broker = CredentialBroker()
broker.register_credential("github_tool", os.environ["GITHUB_TOKEN"])

agent = Agent(model=model, reasoning=ReAct(), tools=tools, credential_broker=broker)
# Tools with requires_credentials=True receive a ScopedToken (repr-safe, auto-expiring)

Observability (OTel + SIEM)

from petfishframework.observability import OTelSink, SIEMSink

# OTel — creates spans for model/tool/session events (requires opentelemetry)
otel_sink = OTelSink()  # no-op if opentelemetry not installed

# SIEM — structured JSON-Lines export for downstream SIEM ingestion
siem_sink = SIEMSink(output_path="audit.jsonl")
# Credentials auto-redacted; configure additional secret fields:
# siem_sink = SIEMSink(output_path="audit.jsonl", redact_keys=("api_key", "password"))

agent = Agent(model=model, reasoning=ReAct(), tools=tools)

# Attach sinks to the session's EventEmitter
session = agent.session("your task")
session.events.subscribe(otel_sink)
session.events.subscribe(siem_sink)
result = session.run()

Tool Governance

from petfishframework.tools import ToolGovernance, ToolSchemaValidator
from petfishframework.tools.rate_limiter import RateLimiter, RateLimitPolicy

governance = ToolGovernance(
    schema_validator=ToolSchemaValidator(),  # validate tool args before execution
    rate_limiter=RateLimiter(),              # per-tool sliding-window rate limit
)

agent = Agent(model=model, reasoning=ReAct(), tools=tools, tool_governance=governance)

Roadmap

• v0.2.x: Core runtime, permission semantics, enterprise PoC, Trusted Publishing ✅
• v0.3.x: YAML Policy Engine, CredentialBroker ✅
• v1.0.x (current): Production release — API stability freeze, thread safety, exception sanitization, Dockerfile hardening, mypy enforcementdes, Vault adapter, Docker, threat model ✅

Current Limitations

petfishFramework is Alpha. API may change before v1.0.

Development

git clone https://github.com/kylecui/petfishFramework.git
cd petfishFramework
uv sync --all-extras
uv run pytest              # 448 tests
uv run ruff check src/ tests/

See CONTRIBUTING.md for details.

License

MIT — © 2026 Kyle Cui