Metadata-Version: 2.4
Name: agents-network
Version: 0.1.0
Summary: A flexible multi-agent orchestration framework with configurable agent and tool registries
Project-URL: Homepage, https://github.com/ginger/agents-network
Project-URL: Repository, https://github.com/ginger/agents-network
Project-URL: Documentation, https://github.com/ginger/agents-network#readme
Project-URL: Issues, https://github.com/ginger/agents-network/issues
Author-email: west <westdynamics.tech@gmail.com>
License: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12
Requires-Dist: httpx>=0.27
Requires-Dist: pydantic>=2.0
Provides-Extra: all
Requires-Dist: mypy>=1.10; extra == 'all'
Requires-Dist: pytest-asyncio>=0.24; extra == 'all'
Requires-Dist: pytest>=8.0; extra == 'all'
Requires-Dist: ruff>=0.5; extra == 'all'
Requires-Dist: streamlit>=1.28; extra == 'all'
Provides-Extra: dev
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.5; extra == 'dev'
Provides-Extra: ui
Requires-Dist: streamlit>=1.28; extra == 'ui'
Description-Content-Type: text/markdown

# agents-network

[![CI](https://github.com/ginger/agents-network/actions/workflows/ci.yml/badge.svg)](https://github.com/ginger/agents-network/actions/workflows/ci.yml)
[![PyPI version](https://img.shields.io/pypi/v/agents-network.svg)](https://pypi.org/project/agents-network/)
[![Python versions](https://img.shields.io/pypi/pyversions/agents-network.svg)](https://pypi.org/project/agents-network/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)

A flexible multi-agent orchestration framework. Agents and tools are discovered through registries, configurable from code, JSON, or any data source.

- **Type-dispatch pattern** — agents/tools registered via decorators, loaded from JSON, DB, or code
- **Semantic matching** — optional embedding-based agent selection via Ollama or OpenAI
- **Audit logging** — tracks selection rates, bias metrics, and human-in-loop approval
- **Guardrails** — configurable blocklist and bias-awareness prompt on built-in agents
- **Orchestrator agent** — delegate routing to an LLM agent
- **Streamlit UI** — dashboard for interactive test-drives

## Installation

```bash
pip install agents-network
```

With extras:

```bash
pip install "agents-network[ui]"   # Streamlit dashboard
pip install "agents-network[all]"  # everything
```

### Using uv

```bash
uv tool install agents-network
# or
uv add agents-network
```

## Quick start

Define a tool and an agent, register them, and let the network route tasks:

```python
import asyncio
from agents_network import Tool, Agent, Task, Message, ToolRegistry, AgentRegistry, AgentNetwork
from agents_network import tool_type, agent_type


@tool_type()
class UppercaseTool(Tool):
    async def run(self, input: dict) -> str:
        return input.get("text", "").upper()


@agent_type()
class TransformerAgent(Agent):
    async def run(self, task: Task, tool_registry: ToolRegistry) -> Message:
        tools = self.select_tools(task, tool_registry)
        result = await tools[0].run({"text": task.input})
        return Message(role="assistant", content=result)


tool_registry = ToolRegistry([UppercaseTool(name="upper", description="converts text to uppercase")])
agent_registry = AgentRegistry([
    TransformerAgent(name="transformer", description="transforms text", tool_names=["upper"]),
])

network = AgentNetwork(agent_registry=agent_registry, tool_registry=tool_registry)

result = asyncio.run(network.run(Task(input="hello world")))
print(result.content)  # "HELLO WORLD"
```

## Config from JSON

Define agents and tools in a JSON file and load at runtime:

```json
{
  "tools": [
    {
      "type": "calculator",
      "name": "calc",
      "description": "Performs arithmetic",
      "precision": 5
    }
  ],
  "agents": [
    {
      "type": "math_agent",
      "name": "math-wizard",
      "description": "Solves math problems",
      "tool_names": ["calc"]
    }
  ]
}
```

```python
network = AgentNetwork.from_json("config.json")
result = asyncio.run(network.run(Task(input="2 + 2")))
```

Each tool and agent references a registered Python class via the `type` field.
Extra fields in the JSON are merged into `config` automatically.

The `plugins` list (or `--load` flag) imports Python modules that register custom types via `@tool_type()` / `@agent_type()` decorators.

```json
{
  "plugins": ["examples.types"],
  "tools": [...],
  "agents": [...]
}
```

Only import modules you trust — `--load` and `plugins` use `importlib.import_module()` and load arbitrary Python code.

### Custom type registration

```python
import ast
import operator

@tool_type("calculator")
class Calculator(Tool):
    precision: int = 2

    _OPS = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
    }

    def _eval(self, node: ast.AST) -> float:
        if isinstance(node, ast.Expression):
            return self._eval(node.body)
        if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
            return float(node.value)
        if isinstance(node, ast.BinOp) and type(node.op) in self._OPS:
            return self._OPS[type(node.op)](self._eval(node.left), self._eval(node.right))
        if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
            return -self._eval(node.operand)
        raise ValueError("Unsupported expression")

    async def run(self, input: dict) -> str:
        tree = ast.parse(input["expression"], mode="eval")
        return str(round(self._eval(tree), self.precision))
```

The name passed to `@tool_type("calculator")` must match the `"type"` field in JSON.

## Agent matching

When a task arrives, `AgentRegistry` scores agents by:

1. **Tag intersection** (3x weight) — agents whose tags overlap task tags
2. **Description keyword match** (1x weight) — words from `agent.description` found in task input

Only agents scoring at or above `min_score` (default: 1) are returned. If no agents meet the threshold, an empty list is returned and the network reports "No suitable agent found".

### Semantic matching

Pass an `EmbeddingMatcher` to `AgentRegistry` for embedding-based scoring:

```python
from agents_network import AgentRegistry, EmbeddingMatcher

matcher = EmbeddingMatcher(model="nomic-embed-text-v2-moe")
registry = AgentRegistry(agents=[...], matcher=matcher)
```

When a matcher is configured, `AgentNetwork.run()` automatically uses `match_semantic()`.

Override for custom routing:

```python
class CustomRegistry(AgentRegistry):
    def match(self, task: Task) -> list[Agent]:
        # your routing logic
        ...
```

## Orchestrator agent

Optionally give the network an orchestrator agent that picks the best match:

```python
@agent_type()
class Router(Agent):
    async def run(self, task: Task, tool_registry: ToolRegistry) -> Message:
        return Message(role="assistant", content="math-wizard")

network = AgentNetwork(
    agent_registry=agent_registry,
    tool_registry=tool_registry,
    orchestrator=Router(name="router", description="routes tasks"),
)
```

## Audit & human-in-loop

```python
from agents_network import AuditLog, AgentNetwork

audit = AuditLog()

async def approve(task):
    # Manual review logic
    return True

network = AgentNetwork(
    agent_registry=agent_registry,
    tool_registry=tool_registry,
    audit_log=audit,
    approval_callback=approve,
)

result = await network.run(task)
print(network.bias_report())  # selection rates per agent
```

## Guardrails

Configure blocked patterns before running ConversationAgent:

```python
from agents_network import set_blocked_patterns

set_blocked_patterns(["ignore all instructions", "role-play"])
```

## CLI

```bash
agents-network "analyse this data" --config config.json
```

| Argument | Default | Description |
|----------|---------|-------------|
| `input` (positional) | — | Task input text |
| `--config` / `-c` | (required) | Path to JSON config |
| `--interactive` / `-i` | `false` | Interactive REPL mode |
| `--load` / `-l` | — | Python module to import for custom types (repeatable) |
| `--tags` / `-t` | `[]` | Tags to attach to the task |

Interactive mode:

```bash
agents-network --config config.json --interactive
```

## Streamlit UI

```bash
pip install "agents-network[ui]"
streamlit run agents_network/ui.py
```

Custom types load via `AGENTS_NETWORK_LOAD` env var:

```bash
AGENTS_NETWORK_LOAD=my_custom_agents streamlit run agents_network/ui.py
```

## DB-backed registries

Subclass `ToolRegistry` or `AgentRegistry` and override:

```python
class PostgresAgentRegistry(AgentRegistry):
    def __init__(self, dsn: str):
        super().__init__()
        rows = query_db(dsn, "SELECT name, description, type, config FROM agents")
        for row in rows:
            self.register(Agent.from_dict(row))
```

## Development

```bash
# Install dev dependencies
uv sync --group dev
# or
pip install "agents-network[dev]"

# Lint
uv run ruff check agents_network/

# Type check
uv run mypy agents_network/

# Test
uv run pytest
```

## CI/CD

Push to any branch triggers:
- `ci.yml` — lint, type-check, test across Python 3.12–3.14, build package

Push a tag (`v*`) triggers:
- `release.yml` — publish to PyPI, create GitHub Release with auto-changelog

## License

MIT
