Metadata-Version: 2.5
Name: agentropic
Version: 0.5.0
Summary: A minimal, hackable agentic framework: LLM calls via litellm, @tool-decorated functions, and MCP server support.
Author-email: Vishal Tiwari <vishaltiwari.up2019@gmail.com>
License: MIT
License-File: LICENSE
Keywords: agent,agents,ai,litellm,llm,mcp,tools
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Requires-Dist: litellm>=1.40.0
Requires-Dist: mcp>=1.0.0; python_version >= '3.10'
Requires-Dist: pydantic>=2.0.0
Provides-Extra: dev
Requires-Dist: build; extra == 'dev'
Requires-Dist: pytest; extra == 'dev'
Requires-Dist: pytest-asyncio; extra == 'dev'
Requires-Dist: twine; extra == 'dev'
Provides-Extra: mcp
Requires-Dist: mcp>=1.0.0; extra == 'mcp'
Description-Content-Type: text/markdown

# agentropic

A small, hackable agentic framework in ~500 lines you fully own and can read
top to bottom in an afternoon:

- **LLM calls** go through [litellm](https://github.com/BerriAI/litellm), so you get
  every provider (OpenAI, Anthropic, Gemini, Ollama, Bedrock, 100+ more) for free.
- **`@tool` decorator** turns any python function into an LLM-callable tool with
  an auto-generated JSON schema (from type hints + docstring).
- **MCP support** — connect to any Model Context Protocol server (stdio or SSE)
  and its tools show up in your agent automatically, indistinguishable from
  local tools.
- **Agent loop** handles the call → tool-call → tool-result → call cycle for you,
  with memory, streaming, and a max-iteration safety valve.

## Install (once published)

```bash
pip install agentropic
```

## Quick start

```python
import asyncio
from agentropic import Agent, tool

@tool()
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"It's sunny in {city}."

async def main():
    agent = Agent(
        name="assistant",
        model="gpt-4o-mini",         # any litellm model string
        instructions="You are helpful and concise.",
        tools=[get_weather],
    )
    print(await agent.arun("What's the weather in Paris?"))

asyncio.run(main())
```

Sync usage: `agent.run("...")` (wraps `arun` in `asyncio.run` for you).

## Streaming (`stream=True`)

Same shape as the OpenAI / Anthropic SDKs. Tool calls work while streaming.

```python
async for event in await agent.arun("Weather in Tokyo?", stream=True):
    if event.type == "text":
        print(event.text, end="", flush=True)
    elif event.type == "tool_call":
        print(f"\n[calling {event.tool_call.name}({event.tool_call.arguments})]")
    elif event.type == "tool_result":
        print(f"[-> {event.tool_call.result}]")
    elif event.type == "done":
        result = event.result          # full AgentRunResult (usage, tool_calls, ...)
```

With `verbose=True` (the default) the colourful terminal tracer already prints
tokens live, so only print `event.text` yourself when using `verbose=False`
(e.g. rendering into your own UI).

Sync: `for event in agent.run("...", stream=True): ...`
Text only: `async for token in agent.astream("..."): ...`

## Human-in-the-loop (tool approvals)

Mark sensitive tools, then choose how approvals happen. The design follows
the OpenAI Agents SDK (interruptions + `RunState`), Pydantic AI (deferred
tools) and the Claude Agent SDK (`can_use_tool` callback).

```python
from agentropic import Agent, tool, ApprovalDecision, ConsoleApprover, RunState

@tool(requires_approval=True)                              # always ask
def delete_file(path: str) -> str: ...

@tool(requires_approval=lambda args: args["amount"] > 1000)  # ask only sometimes
def refund(order_id: str, amount: float) -> str: ...
```

**Mode A — inline handler** (CLIs, scripts). The run waits for your callback:

```python
agent = Agent(..., tools=[delete_file], approval_handler=ConsoleApprover())
# ConsoleApprover asks: [y]es / [n]o / [a]lways / [e]dit args

# or your own (sync or async), returning bool or ApprovalDecision:
async def my_handler(req):            # req.tool_name, req.arguments, req.id
    return ApprovalDecision.deny("Not during business hours")
```

**Mode B — pause & resume** (web apps, Slack/email approvals). With no
handler, the run stops and returns a serializable state:

```python
result = await agent.arun("Clean up old files")
while result.is_interrupted:
    state = result.to_state()
    db.save(state.to_json())                 # optional: resume later / elsewhere

    state = RunState.from_json(db.load())
    for req in state.pending:
        state.approve(req)                   # approve(req, override_args={...}, always=True)
        # state.reject(req, message="Too risky")
    result = await agent.arun(state)         # or arun(state, stream=True)
print(result)
```

Details:
- A denial is sent back to the model as the tool result, so it can adapt.
- Tools that don't need approval in the same turn run before the pause.
- `always=True` remembers the decision for that tool for the rest of the run.
- Asking a new question while approvals are pending auto-cancels them.
- Policies fail closed: if the model sends unparsable arguments, or your
  policy function raises, approval is required.
- Streaming emits an `approval_required` event.
- MCP servers: `StdioServer(..., require_approval="always" | {"write_file"} | fn(name, args))`.
- `RunState.to_json()` contains the full conversation; store it like user data.

## Multiple agents

Two patterns, modelled on the OpenAI Agents SDK. Give each agent a short
`description`; other agents see it when deciding whom to call.

**Agents as tools**: a manager calls specialists like functions and
combines their answers. Each call is an isolated run with the sub-agent's
own tools and instructions.

```python
researcher = Agent(name="researcher", model=..., description="Finds facts.", tools=[search])
writer     = Agent(name="writer", model=..., description="Writes polished copy.")

manager = Agent(
    name="manager", model=...,
    instructions="Research first, then have the writer draft the answer.",
    tools=[researcher.as_tool(), writer.as_tool()],
    # as_tool(name=..., description=..., requires_approval=..., persist_memory=False)
)
print(await manager.arun("Write a short brief on solid-state batteries"))
```

**Handoffs**: the model transfers the whole conversation to a specialist,
which then answers the user directly (a `transfer_to_<name>` tool is added
automatically).

```python
triage = Agent(name="triage", model=..., handoffs=[billing, tech_support])
result = await triage.arun("I was charged twice")
result.last_agent      # -> billing
result.agent_path      # -> ["triage", "billing"]
await result.last_agent.arun("When will I get the refund?")   # keep chatting
```

How the two features work together:
- **Approvals from sub-agents:** if a sub-agent needs approval, it uses the
  calling agent's `approval_handler`. With no handler, the outer run pauses,
  and `result.interruptions` shows each request with `agent_name` and
  `parent_call_id`. Approve or reject on the outer state and resume the
  outer agent.
- **Resuming after a handoff:** if a specialist pauses after a handoff,
  `triage.arun(state)` routes the resume to the specialist automatically.
- **Loop protection:** `max_handoffs` (default 5) stops agents from
  passing the conversation back and forth forever.
- **Memory:** a handoff replaces the target agent's memory with the
  current conversation. Use one agent set per user or session.
- **Streaming:** a `handoff` event is emitted, and the terminal tracer
  shows `🔀 Handoff: triage → billing`.

## Using MCP servers

```python
from agentropic import Agent
from agentropic.mcp import StdioServer, HTTPServer

agent = Agent(
    name="fs-agent",
    model="gpt-4o-mini",
    mcp_servers=[
        StdioServer(command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", "."]),
        # HTTPServer(url="https://example.com/mcp", headers={"Authorization": "Bearer ..."}),
    ],
)

# Option A: connect once, keep servers alive across turns (recommended)
async with agent:
    print(await agent.arun("List the files in the current directory."))
    print(await agent.arun("Now read README.md"))

# Option B: no context manager — servers connect/disconnect around each run
print(await agent.arun("List the files in the current directory."))
```

Requires the optional MCP dependency: `pip install "agentropic[mcp]"`, plus whatever
the server itself needs (e.g. Node.js for `npx` servers). Transports: `StdioServer`,
`HTTPServer` (streamable HTTP), `SSEServer` (legacy SSE).

## Architecture

```
src/agentropic/
├── __init__.py     # public API surface
├── agent.py         # the run loop: LLM <-> tools <-> memory
├── llm.py            # thin litellm wrapper (swap providers here)
├── tools.py           # @tool decorator, Tool, ToolRegistry, schema generation
├── memory.py           # in-memory conversation history (swap for your own store)
└── mcp/
    └── client.py        # MCP stdio/SSE client -> Tool adapter
```

Everything is a plain, editable python file — no hidden magic, no plugin
registry you can't inspect. Read `agent.py` first; that's the whole loop.

## Extending it

- **Custom memory / persistence**: implement `.add()`, `.get()`, `.clear()`
  matching `Memory` and pass `memory=YourMemory()` to `Agent`.
- **Multi-agent handoff**: compose multiple `Agent` instances and route
  between them yourself, or wrap one agent's `arun` as a tool for another.

