Metadata-Version: 2.4
Name: fluxos-agent
Version: 2.0.1
Summary: Stateful agent runtime with tool execution and event streaming
Author-email: fluxos contributors <dev@fluxtopus.com>
License: MIT
Classifier: Development Status :: 4 - Beta
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.27.0
Provides-Extra: dev
Requires-Dist: pytest>=7.4.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: pytest-cov>=4.1.0; extra == "dev"
Requires-Dist: ruff>=0.1.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"

# fluxos-agent

Python port of `pi-mono/packages/agent` [https://github.com/badlogic/pi-mono/tree/main/packages/agent].

`fluxos-agent` provides:
- Stateful agent runtime (`Agent`) with prompt/continue flows
- Agent loop primitives (`agent_loop`, `agent_loop_continue`)
- Tool execution with streaming update events
- Steering and follow-up queues
- Optional proxy stream adapter (`stream_proxy`)

## Installation

Install from PyPI (recommended):

```bash
uv add fluxos-agent
# or: pip install fluxos-agent
```

Local editable install (repo development):

```bash
pip install -e packages/fluxos-agent
```

## Quick Start

```python
import asyncio

from flux_agent import Agent, AgentOptions
from flux_agent.types import (
    AssistantDoneEvent,
    AssistantMessage,
    Context,
    Model,
    TextContent,
    Usage,
    UsageCost,
)
from flux_agent.event_stream import EventStream


class MockAssistantStream(EventStream):
    def __init__(self):
        super().__init__(
            lambda event: event.type in {"done", "error"},
            lambda event: event.message if event.type == "done" else event.error,
        )


def stream_fn(model: Model, context: Context, options):
    stream = MockAssistantStream()
    message = AssistantMessage(
        content=[TextContent(text="Hello from Python agent")],
        api=model.api,
        provider=model.provider,
        model=model.id,
        usage=Usage(cost=UsageCost()),
        stop_reason="stop",
    )
    stream.push(AssistantDoneEvent(reason="stop", message=message))
    return stream


async def main():
    agent = Agent(
        AgentOptions(
            stream_fn=stream_fn,
            initial_state={
                "system_prompt": "You are helpful.",
                "model": Model(id="mock", name="mock", api="mock", provider="mock"),
            },
        )
    )

    agent.subscribe(lambda event: print(event.type))
    await agent.prompt("Hello")


asyncio.run(main())
```

## Testing

```bash
cd packages/fluxos-agent
pip install -e ".[dev]"
pytest tests/unit -v
```

Run full suite (unit + e2e scaffolding):

```bash
pytest tests -v
```

## Release Tracking

- Keep package changes tracked in `CHANGELOG.md`.
- For every release:
  1. Update `pyproject.toml` version.
  2. Add a new dated section in `CHANGELOG.md`.
  3. Publish and tag the release.
