fabrix-ai 1.0.0

Graph-based agent framework powered by oauth-codex

Fabrix

Language: English | 한국어
API Guides: English | 한국어

Overview

Fabrix is a graph-based agent framework built on top of oauth-codex>=2.2.0. It provides a structured execution graph with streaming events for tool-driven workflows.

Key Features

• Graph-based 4-state execution: reasoning, tool_call, response, finish
• Structured state outputs powered by Pydantic models
• Sequential tool execution with strict payload validation
• Async streaming event API for step-by-step observability

Installation

pip install fabrix-ai

Quickstart

import asyncio

from pydantic import BaseModel

from fabrix import Agent
from fabrix.events import (
    ReasoningEvent,
    ResponseEvent,
    TaskFailedEvent,
    TaskFinishedEvent,
    ToolEvent,
)


class AddInput(BaseModel):
    a: int
    b: int


def add_numbers(payload: AddInput) -> int:
    return payload.a + payload.b


async def main() -> None:
    agent = Agent(
        instructions="You are a precise assistant.",
        model="gpt-5.3-codex",
        tools=[add_numbers],
    )

    async for event in agent.run_task_stream("Use add_numbers to compute 3 + 9"):
        print(f"[step={event.step}] {event.event_type}")

        if isinstance(event, ReasoningEvent):
            print("reasoning:", event.reasoning)
            print("focus:", event.focus)
        elif isinstance(event, ToolEvent):
            if event.phase == "start":
                print("tool call:", event.tool_name, event.arguments)
            elif event.result is not None:
                print("tool result:", event.result.model_dump())
        elif isinstance(event, ResponseEvent):
            print("response:", event.response)
        elif isinstance(event, TaskFinishedEvent):
            print("completion reason:", event.completion_reason)
            print("final:", event.final_output)
        elif isinstance(event, TaskFailedEvent):
            print("failed:", event.error_code, event.message)


asyncio.run(main())

Tool Contract

Fabrix accepts tools in this shape:

def tool(payload: BaseModel) -> Any: ...

• The tool must accept exactly one parameter.
• The parameter type must be a Pydantic BaseModel.
• Runtime arguments must be a JSON object matching payload fields.
• Extra argument keys are rejected.
• Both sync and async tools are supported.

Event Stream

run_task_stream(...) yields these event types:

• reasoning
• tool (phase="start" / phase="finish")
• response
• task_finished
• task_failed

reasoning is a step-level decision trace / plan summary, not raw internal chain-of-thought.

Typical flow:

1. reasoning
2. zero or more tool start/finish pairs
3. optional response
4. task_finished or task_failed

Documentation

• API usage guide (English): docs/api.md
• API 사용 가이드 (한국어): docs/api.ko.md
• Korean README: README.ko.md

Examples

• Minimal quickstart: examples/minimal/quickstart.py
• Data workflow: examples/advanced/data_workflow.py
• Incident response workflow: examples/advanced/incident_response.py

Notes

• Public runtime entry point is fabrix.Agent.
• Execution defaults are fixed internally: max_steps=128 and no public per-tool timeout option.
• If max_steps is reached and at least one response was emitted, the stream ends with task_finished (completion_reason="max_steps_reached") using the last response.
• If max_steps is reached before any response/final output, the stream ends with task_failed (error_code="max_steps_reached").