agents-network

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

pip install agents-network

With extras:

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

Using uv

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:

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:

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

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.

{
  "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

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:

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:

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:

@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

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:

from agents_network import set_blocked_patterns

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

CLI

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

Interactive mode:

agents-network --config config.json --interactive

Streamlit UI

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

Custom types load via AGENTS_NETWORK_LOAD env var:

AGENTS_NETWORK_LOAD=my_custom_agents streamlit run agents_network/ui.py

DB-backed registries

Subclass ToolRegistry or AgentRegistry and override:

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

# 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