agentgrid-py 0.1.0

Task routing layer for production AI agents — dispatch to the right agent by capability, health, and policy

agentgrid-py

Task routing for production AI agents. Given a task, AgentGrid finds the right agent from your registry based on capabilities, health, and policy — then dispatches.

Install

pip install agentgrid-py
pip install "agentgrid-py[registry]"   # with agentregistry-py
pip install "agentgrid-py[all]"        # with registry + mesh + policy

Quickstart

import asyncio
from agentgrid import AgentGrid
from agentregistry import AgentRegistry, AgentManifest

async def main():
    registry = AgentRegistry()
    registry.publish(AgentManifest(
        id="acme.search-agent", version="1.0.0",
        capabilities=["search", "summarize"],
        framework="langgraph",
    ))

    grid = AgentGrid(registry=registry)

    # Register handler for local dispatch
    async def search_handler(task: str, payload: dict) -> dict:
        return {"output": f"search results for: {task}"}
    grid.register("acme.search-agent", search_handler)

    # Route task to best agent
    result = await grid.route("find quarterly revenue", requires=["search"])
    print(f"Routed to: {result.agent_id} (score={result.score:.2f})")
    print(f"Response: {result.response}")

asyncio.run(main())

Routing Strategies

from agentgrid import RoutingStrategy

# Best capability match (default)
result = await grid.route(task, strategy=RoutingStrategy.BEST_MATCH)

# Round-robin across eligible agents
result = await grid.route(task, strategy=RoutingStrategy.ROUND_ROBIN)

# Least loaded agent
result = await grid.route(task, strategy=RoutingStrategy.LEAST_LOADED)

# Broadcast to all eligible agents
results = await grid.broadcast(task, requires=["handle_event"])

Stack

agentgrid    → routing       which agent handles what task
agentmesh    → events        connects agents via pub/sub
agentplane   → policy        governs agent behaviour
agentregistry → discovery    what agents exist and their capabilities

---

Apache 2.0 · PyPI