Organize reasoners and skills within an Agent Node using modular routing patterns

Code organization tool for grouping related reasoners and skills within a single Agent Node. Provides FastAPI-style modular routing with automatic prefix-based namespacing.

What It Is

AgentRouter organizes functions within an Agent Node, not across nodes. Think of it like Express.js routers or FastAPI sub-applications—a way to split large API files into logical modules that you import and compose in main.py.

Architectural Distinction:

Agent Node = Microservice (distributed deployment, own DID, cross-network calls via app.call())
Router = Code organization within one Agent Node (modular files, namespace management, team collaboration)

If you need distributed communication, create separate Agent Nodes. If you need code organization, use Routers.

How Router Prefixes Affect API Calls

This is the key concept: When you organize functions with AgentRouter, the router's prefix becomes part of the API endpoint. This is how you namespace your agent's functions and avoid collisions.

Quick Example

# routers/user_management.py
users = AgentRouter(prefix="users")

@users.reasoner()
async def analyze_user_behavior(user_id: int) -> dict:
    """AI-powered user behavior analysis."""
    return await users.ai(...)

@users.skill(tags=["database"])
def get_user_profile(user_id: int) -> dict:
    """Retrieve user profile from database."""
    return database.get_user(user_id)

# routers/analytics.py
analytics = AgentRouter(prefix="analytics")

@analytics.reasoner()
async def generate_insights(data: dict) -> dict:
    """Generate business insights from data."""
    return await analytics.ai(...)

# main.py
app.include_router(users)      # users_analyze_user_behavior, users_get_user_profile
app.include_router(analytics)  # analytics_generate_insights

Calling Router Functions via HTTP

The router prefix translates directly to the API endpoint:

# Calling the users router reasoner
curl -X POST http://localhost:8080/api/v1/execute/user-agent.users_analyze_user_behavior \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "user_id": "abc123"
    }
  }'

# Calling the users router skill
curl -X POST http://localhost:8080/api/v1/execute/user-agent.users_get_user_profile \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "user_id": "abc123"
    }
  }'

# Calling the analytics router reasoner
curl -X POST http://localhost:8080/api/v1/execute/user-agent.analytics_generate_insights \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "data": {"metric": "engagement", "period": "30d"}
    }
  }'

Prefix Translation Rules

Router prefixes are automatically converted to valid identifiers:

Router Prefix	Function Name	API Endpoint
`"billing"`	`calculate_cost`	`agent.billing_calculate_cost`
`"Support/Inbox"`	`route_ticket`	`agent.support_inbox_route_ticket`
`"API/v2/Users"`	`create_user`	`agent.api_v2_users_create_user`
`"ML-Models/GPT-4"`	`generate_text`	`agent.ml_models_gpt_4_generate_text`

Translation algorithm:

Lowercase the prefix
Replace non-alphanumeric characters with underscores
Join segments with underscores
Prepend to function name

Without a router: Function analyze_sentiment becomes agent.analyze_sentiment

With router prefix "users": Same function becomes agent.users_analyze_sentiment

This is how you organize large agents and avoid naming collisions.

Custom Name Override

When you need a stable identifier (e.g., for legacy systems), override the generated name:

router = AgentRouter(prefix="Users/Profile-v1")

# Generated ID: users_profile_v1_get_profile
@router.reasoner()
async def get_profile(user_id: str) -> dict:
    return {"user_id": user_id}

# Custom ID: accounts_sync_profile (ignores prefix)
@router.reasoner(name="accounts_sync_profile")
async def sync_profile(user_id: str) -> dict:
    return {"synced": True}

API calls:

# Using generated ID
curl -X POST http://localhost:8080/api/v1/execute/user-agent.users_profile_v1_get_profile \
  -d '{"input": {"user_id": "abc123"}}'

# Using custom name override
curl -X POST http://localhost:8080/api/v1/execute/user-agent.accounts_sync_profile \
  -d '{"input": {"user_id": "abc123"}}'

Basic Example

# routers/billing.py
from agentfield.router import AgentRouter

billing = AgentRouter(prefix="billing")

@billing.reasoner()
async def calculate_invoice(customer_id: int, month: str) -> dict:
    """Generate monthly invoice with AI-powered cost optimization."""
    return await billing.ai(
        system="Calculate invoice with cost optimization suggestions",
        user=f"Customer {customer_id}, month {month}",
        schema=InvoiceResult
    )

@billing.skill(tags=["payment"])
def process_payment(invoice_id: str, amount: float) -> dict:
    """Process payment transaction."""
    return {"invoice_id": invoice_id, "status": "processed"}

# routers/support.py
from agentfield.router import AgentRouter

support = AgentRouter(prefix="support")

@support.reasoner()
async def route_ticket(ticket_text: str) -> dict:
    """Route support ticket to appropriate team."""
    return await support.ai(
        system="Classify support ticket urgency and department",
        user=ticket_text,
        schema=TicketRouting
    )

@support.skill(tags=["notification"])
def send_ticket_notification(ticket_id: str, team: str) -> dict:
    """Send notification to assigned team."""
    return {"ticket_id": ticket_id, "notified": team}

# main.py
from agentfield import Agent
from routers.billing import billing
from routers.support import support

app = Agent(node_id="customer_service")

# Include routers - prefixes become part of function IDs
app.include_router(billing)   # billing_calculate_invoice, billing_process_payment
app.include_router(support)   # support_route_ticket, support_send_ticket_notification

if __name__ == "__main__":
    app.serve()

# Call billing reasoner
curl -X POST http://localhost:8080/api/v1/execute/customer_service.billing_calculate_invoice \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "customer_id": 12345,
      "month": "2024-07"
    }
  }'

# Call support reasoner
curl -X POST http://localhost:8080/api/v1/execute/customer_service.support_route_ticket \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "ticket_text": "My payment failed, need urgent help"
    }
  }'

# Cross-agent call from another agent
result = await app.call(
    "customer_service.billing_calculate_invoice",
    customer_id=12345,
    month="2024-07"
)

When to Use Routers

Decision Criteria:

Agent Size	Recommendation	Reason
< 10 functions	No router needed	Direct decorators in `main.py` are clearer
10-30 functions	Consider routers	Improves readability and navigation
30+ functions	Routers essential	Prevents monolithic files, enables team collaboration
Team collaboration	Always use routers	Each team owns their router module
API versioning	Always use routers	Separate v1, v2 routers for clean migration

✅ Use Routers When

Organizing large agents with many functions
Multiple developers working on the same agent
Versioning API endpoints (v1, v2, v3)
Grouping by feature domain (billing, support, analytics)
Avoiding namespace collisions in complex agents
Splitting code across multiple files for maintainability

❌ Don't Use Routers When

Building simple agents with < 10 functions
Need distributed communication (use separate Agent Nodes instead)
Prototyping or experimenting (add routers later when needed)

Common Patterns

Feature-Based Organization

# routers/user_management.py
users = AgentRouter(prefix="users")

@users.reasoner()
async def analyze_user_behavior(user_id: int) -> dict:
    """AI-powered user behavior analysis."""
    return await users.ai(...)

@users.skill(tags=["database"])
def get_user_profile(user_id: int) -> dict:
    """Retrieve user profile from database."""
    return database.get_user(user_id)

# routers/analytics.py
analytics = AgentRouter(prefix="analytics")

@analytics.reasoner()
async def generate_insights(data: dict) -> dict:
    """Generate business insights from data."""
    return await analytics.ai(...)

# main.py
app.include_router(users)      # users_analyze_user_behavior, users_get_user_profile
app.include_router(analytics)  # analytics_generate_insights

API Versioning

# routers/api_v1.py
v1 = AgentRouter(prefix="api/v1")

@v1.reasoner()
async def classify_content(text: str) -> dict:
    """Legacy classification (deprecated)."""
    return await v1.ai(...)

# routers/api_v2.py
v2 = AgentRouter(prefix="api/v2")

@v2.reasoner()
async def classify_content(text: str, model: str = "gpt-4") -> dict:
    """Enhanced classification with model selection."""
    return await v2.ai(model=model, ...)

# main.py
app.include_router(v1)  # api_v1_classify_content
app.include_router(v2)  # api_v2_classify_content

# Clients can migrate gradually:
# Old: customer_service.api_v1_classify_content
# New: customer_service.api_v2_classify_content

Team Ownership

# routers/team_billing.py (owned by billing team)
billing_team = AgentRouter(prefix="billing", tags=["billing-team"])

@billing_team.reasoner()
async def calculate_costs(usage_data: dict) -> dict:
    return await billing_team.ai(...)

# routers/team_support.py (owned by support team)
support_team = AgentRouter(prefix="support", tags=["support-team"])

@support_team.reasoner()
async def handle_escalation(ticket: dict) -> dict:
    return await support_team.ai(...)

# main.py - each team maintains their router independently
app.include_router(billing_team)
app.include_router(support_team)

Nested Namespaces

# Complex namespace hierarchy
admin = AgentRouter(prefix="admin/users/v2")

@admin.reasoner()
async def audit_user_activity(user_id: int) -> dict:
    return await admin.ai(...)

app.include_router(admin)
# Registered as: admin_users_v2_audit_user_activity
# API: /api/v1/execute/agent.admin_users_v2_audit_user_activity

Constructor Parameters

Prop

Type

# Basic router
router = AgentRouter(prefix="billing")

# Router with default tags
router = AgentRouter(
    prefix="support/v2",
    tags=["support-team", "production"]
)

# No prefix (functions use their original names)
router = AgentRouter()

Decorator Methods

`@router.reasoner()`

Prop

Type

router = AgentRouter(prefix="analytics")

# Default path: /reasoners/analytics_analyze_data
@router.reasoner()
async def analyze_data(data: dict) -> dict:
    return await router.ai(...)

# Custom path
@router.reasoner(path="/custom/analysis")
async def custom_analysis(data: dict) -> dict:
    return await router.ai(...)

# Override generated ID
@router.reasoner(name="legacy_analyzer")
async def analyze_legacy(data: dict) -> dict:
    return await router.ai(...)

`@router.skill()`

Prop

Type

router = AgentRouter(prefix="billing", tags=["billing-team"])

# Inherits router tags: ["billing-team"]
@router.skill()
def calculate_tax(amount: float) -> float:
    return amount * 0.08

# Merges tags: ["billing-team", "payment", "critical"]
@router.skill(tags=["payment", "critical"])
def process_payment(invoice_id: str) -> dict:
    return {"status": "processed"}

# Custom path and name
@router.skill(path="/legacy/refund", name="legacy_refund_processor")
def process_refund(amount: float) -> dict:
    return {"refunded": amount}

Delegation Methods

Routers provide access to agent methods after attachment via include_router().

`router.ai()`

Access the agent's AI interface from within router functions.

router = AgentRouter(prefix="content")

@router.reasoner()
async def analyze_content(text: str) -> dict:
    # Use router.ai() instead of app.ai()
    result = await router.ai(
        system="Analyze content for sentiment and topics",
        user=text,
        schema=ContentAnalysis
    )
    return result

`router.call()`

Make cross-agent calls from within router functions.

router = AgentRouter(prefix="orchestrator")

@router.reasoner()
async def orchestrate_workflow(user_id: int) -> dict:
    # Call other agents via router.call()
    user_data = await router.call(
        "user_service.get_user_profile",
        user_id=user_id
    )

    analysis = await router.call(
        "analytics_service.analyze_user",
        user_data=user_data
    )

    return {"user_data": user_data, "analysis": analysis}

`router.memory`

Access the agent's memory system from within router functions.

router = AgentRouter(prefix="session")

@router.reasoner()
async def process_session(session_id: str, data: dict) -> dict:
    # Store session data in memory
    await router.memory.set(f"session_{session_id}", data)

    # Retrieve previous session data
    previous = await router.memory.get(f"session_{session_id}_previous")

    return {"current": data, "previous": previous}

Router delegation methods (ai, call, memory) only work after the router is attached to an agent via app.include_router(). Calling them before attachment raises RuntimeError.

Prefix Translation Rules

Router prefixes are automatically translated to valid Python identifiers for function registration.

Translation Algorithm

Strip leading/trailing slashes
Split on / to get segments
For each segment:
- Replace non-alphanumeric characters with _
- Collapse multiple _ to single _
- Strip leading/trailing _
- Convert to lowercase
Join segments with _
Append _ to create prefix

Translation Examples

Router Prefix	Function Name	Registered ID	API Endpoint
`"billing"`	`calculate_cost`	`billing_calculate_cost`	`agent.billing_calculate_cost`
`"Billing"`	`calculate_cost`	`billing_calculate_cost`	`agent.billing_calculate_cost`
`"Support/Inbox"`	`route_ticket`	`support_inbox_route_ticket`	`agent.support_inbox_route_ticket`
`"API/v2/Users"`	`create_user`	`api_v2_users_create_user`	`agent.api_v2_users_create_user`
`"ML-Models/GPT-4"`	`generate`	`ml_models_gpt_4_generate`	`agent.ml_models_gpt_4_generate`
`"Users/Profile-v1"`	`get_profile`	`users_profile_v1_get_profile`	`agent.users_profile_v1_get_profile`
`""` (empty)	`my_function`	`my_function`	`agent.my_function`

Explicit Name Override

Override the generated ID when you need stable identifiers for legacy systems or migrations.

router = AgentRouter(prefix="complex/namespace/v2")

# Generated ID: complex_namespace_v2_process_data
@router.reasoner()
async def process_data(data: dict) -> dict:
    return await router.ai(...)

# Explicit ID: stable_processor (ignores prefix)
@router.reasoner(name="stable_processor")
async def process_data_v2(data: dict) -> dict:
    return await router.ai(...)

# Cross-agent calls use the explicit name
result = await app.call("agent.stable_processor", data={...})

Including Routers

Use app.include_router() to attach routers to an Agent Node.

Basic Inclusion

from agentfield import Agent
from routers.billing import billing
from routers.support import support

app = Agent(node_id="customer_service")

# Include routers - functions are registered with prefixed IDs
app.include_router(billing)
app.include_router(support)

Nested Prefixes

Add additional prefix when including the router.

# Router has prefix "users"
users = AgentRouter(prefix="users")

@users.reasoner()
async def get_profile(user_id: int) -> dict:
    return await users.ai(...)

# Include with additional prefix "api/v2"
app.include_router(users, prefix="api/v2")

# Final registered ID: api_v2_users_get_profile
# API: /api/v1/execute/agent.api_v2_users_get_profile

Tag Merging

Tags from include_router() are merged with router and skill tags.

router = AgentRouter(prefix="billing", tags=["billing-team"])

@router.skill(tags=["payment"])
def process_payment(amount: float) -> dict:
    return {"processed": amount}

# Include with additional tags
app.include_router(router, tags=["production", "critical"])

# Final tags: ["production", "critical", "billing-team", "payment"]

Router vs Agent Node

Understanding when to use routers versus separate agent nodes is crucial for proper architecture.

Aspect	Router	Agent Node
Purpose	Code organization	Distributed microservice
Deployment	Part of parent agent	Independent deployment
Communication	Direct function calls	`app.call()` via Agentfield server
Identity	No separate DID	Own DID and credentials
Scaling	Scales with parent agent	Independent scaling
Team Ownership	Module within agent	Separate service ownership
Use Case	Organize 30+ functions	Separate business domain

Example Architecture:

# ❌ WRONG: Using routers for distributed concerns
app = Agent(node_id="monolith")
app.include_router(billing_router)      # Should be separate agent
app.include_router(user_router)         # Should be separate agent
app.include_router(notification_router) # Should be separate agent

# ✅ CORRECT: Separate agents for distributed concerns
billing_agent = Agent(node_id="billing_service")
user_agent = Agent(node_id="user_service")
notification_agent = Agent(node_id="notification_service")

# ✅ CORRECT: Routers for organizing within an agent
billing_agent = Agent(node_id="billing_service")
billing_agent.include_router(invoicing_router)  # Organize invoicing functions
billing_agent.include_router(payments_router)   # Organize payment functions
billing_agent.include_router(refunds_router)    # Organize refund functions