Tools

Copy Page

Tools are interfaces that allow models to perform actions or query information.

What are Tools?

1. Function Calling - A tool is exposed as a function with defined name, parameters, and types

   • Example: web_search(query: str)
   • The model decides whether to call it and provides arguments

2. Extending Capabilities - Tools allow you to:

   • Search for real-time data (news, stocks, databases)
   • Execute Code
   • Manipulate systems (send emails, schedule events)
   • Integrate with external APIs

3. Agent-based Orchestration - The LLM acts as an agent that decides:

   • When to use a tool
   • Which tool to use
   • How to interpret the tool's output

In msgFlux, a Tool can be any callable (function, class with __call__/acall e.g. nn.Agent).

Info

While more tools enable more actions, too many tools can confuse the model about which one to use.

Tip

A good practice is to inform the model in the system prompt when it should use that tool.

How Tool Calls Work

When the model decides to use a tool, the Agent intercepts the response, executes the function, appends the result to the conversation, and calls the model again. This loop continues until the model produces a final text response.

                        Input
                          │
                          ▼
┌──────────────────────────────────────────────┐
│            messages + tool schemas           │
└─────────────────────────┬────────────────────┘
                          │
                          ▼
                  ┌───────────────┐
                  │     Model     │ ──▶ "call get_weather(city)"
                  └───────────────┘
                          │
                          ▼
                  ┌───────────────┐
                  │  get_weather  │ ──▶ "Sunny, 24°C"
                  └───────────────┘
                          │  result appended to messages
                          ▼
                  ┌───────────────┐
                  │     Model     │ ──▶ "The weather in Paris is sunny..."
                  └───────┬───────┘
                          │
                     More calls?
                    /            \
                  Yes             No
                   │               │
             (next cycle)     [ Output ]

Example

GitHub APIFile OperationsWeb FetchWikipedia Search

Query GitHub's public API for repository information:

# pip install msgflux[openai]
import httpx
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

def get_github_repo(owner: str, repo: str) -> str:
    """Get information about a GitHub repository.

    Args:
        owner: Repository owner (username or organization).
        repo: Repository name.

    Returns:
        Repository details including stars, forks, and description.
    """
    url = f"https://api.github.com/repos/{owner}/{repo}"
    response = httpx.get(url, timeout=10)

    if response.status_code == 404:
        return f"Repository {owner}/{repo} not found."

    if response.status_code != 200:
        return f"Error fetching repository: {response.status_code}"

    data = response.json()
    return f"""
    Repository: {data['full_name']}
    Description: {data.get('description', 'No description')}
    Stars: {data['stargazers_count']:,}
    Forks: {data['forks_count']:,}
    Language: {data.get('language', 'Unknown')}
    Open Issues: {data['open_issues_count']}
    URL: {data['html_url']}
    """

class GithubAssistant(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    system_message = "You help users explore GitHub repositories."
    tools = [get_github_repo]
    config = {"verbose": True}

response = agent("Tell me about the pytorch repository")

Real file system operation:

# pip install msgflux[openai]
import os
from pathlib import Path
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

def list_files(directory: str, pattern: str = "*") -> str:
    """List files in a directory matching a pattern.

    Args:
        directory: Path to the directory.
        pattern: Glob pattern to filter files (default: all files).

    Returns:
        List of matching files with sizes.
    """
    path = Path(directory).expanduser()

    if not path.exists():
        return f"Directory not found: {directory}"

    if not path.is_dir():
        return f"Not a directory: {directory}"

    files = list(path.glob(pattern))[:20]  # Limit results

    if not files:
        return f"No files matching '{pattern}' in {directory}"

    result = []
    for f in files:
        size = f.stat().st_size if f.is_file() else 0
        size_str = f"{size:,} bytes" if f.is_file() else "directory"
        result.append(f"  {f.name} ({size_str})")

    return f"Files in {directory}:\n" + "\n".join(result)

def read_file(filepath: str, max_lines: int = 50) -> str:
    """Read content from a text file.

    Args:
        filepath: Path to the file.
        max_lines: Maximum lines to read (default: 50).

    Returns:
        File content or error message.
    """
    path = Path(filepath).expanduser()

    if not path.exists():
        return f"File not found: {filepath}"

    if not path.is_file():
        return f"Not a file: {filepath}"

    try:
        lines = path.read_text().splitlines()[:max_lines]
        content = "\n".join(lines)
        if len(lines) == max_lines:
            content += "\n...[truncated]"
        return content
    except Exception as e:
        return f"Error reading file: {e}"

class FileAssistant(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    system_message = "You help users explore files on their system."
    tools = [list_files, read_file]
    config = {"verbose": True}

response = agent("List Python files in the current directory")

Extract text content from web pages using httpx:

# pip install msgflux[openai] beautifulsoup4
import msgflux as mf
import msgflux.nn as nn
from msgflux.tools.builtin import WebFetch

# mf.set_envs(OPENAI_API_KEY="...")

class WebReader(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    system_message = "You help users understand web content."
    tools = [WebFetch]
    config = {"verbose": True}

agent = WebReader()

response = agent("Summarize the main points from https://news.ycombinator.com")

Use msgflux's built-in Wikipedia retriever as a tool:

# pip install msgflux[openai] wikipedia
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

# Create Wikipedia search tool from built-in retriever
wikipedia = mf.Retriever.web_search("wikipedia")

class Researcher(nn.Module):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    system_message = "You are a research assistant with access to Wikipedia.",
    tools = [wikipedia]
    config = {"verbose": True}

response = agent("Tell me about the history of the Python programming language")

Writing Good Tools

Tool Names

A well-defined tool is fundamental for the model to understand when and how to use it. The model reads the tool's name, description (docstring), and parameter definitions to decide if it should call the tool and with what arguments.

Poor tool definitions lead to:

• The model not calling the tool when it should
• Incorrect parameter values being passed
• Confusion when multiple tools have similar names

Tool Names and Description

Good Tool NameBad Tool Name

A simple, descriptive name helps the model quickly understand the tool's purpose. Combined with a clear docstring and well-documented parameters, the model can make accurate decisions about when to use the tool.

Best practices:

• Use short, action-oriented names (search, send_email)
• Document the purpose in the docstring
• Describe each parameter with type hints and descriptions

def web_search(query: str) -> str:
    """Search for content similar to query.

    Args:
        query: Term to search on the web.

    Returns:
        Results similar to query.
    """
    pass

Long, complex names with unnecessary prefixes confuse the model. Missing or poor descriptions make it impossible for the model to understand when to use the tool.

Common problems:

• Overly long names with implementation details (superfast_brave_web_search)
• Redundant parameter names (query_to_search instead of query)
• Missing docstrings or parameter descriptions
• No type hints

def superfast_brave_web_search(query_to_search: str) -> str:
    pass  # No docstring, no parameter description

Tool Returns

The way a tool returns information affects how well the model interprets and uses the result.

Return Value Best Practices

Basic ReturnDescriptive ReturnInstructive Return

Returns the value, but model must infer context:

def add(a: float, b: float) -> float:
    """Sum two numbers."""
    return a + b  # Returns: 8

Provides context that helps the model respond naturally:

def add(a: float, b: float) -> str:
    """Sum two numbers."""
    c = a + b
    return f"The sum of {a} plus {b} is {c}"

Guides the model on how to use the result:

def add(a: float, b: float) -> str:
    """Sum two numbers."""
    c = a + b
    return f"The calculation is complete. Tell the user: {a} + {b} = {c}"

Tool Choice

Control how the model selects tools.

Options:

Value	Behavior
"auto"	Model decides whether to use tools (default)
"required"	Model must call at least one tool
"none"	Model cannot use tools
"tool_name"	Model must call the specific tool

Example

auto (default)requiredSpecific ToolnoneRouter Pattern

Model decides when to use tools:

# pip install msgflux[openai] wikipedia
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

wikipedia = mf.Retriever.web_search("wikipedia")

class Researcher(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    tools = [wikipedia]
    config = {"tool_choice": "auto", "verbose": True}

agent = Researcher()

# Model may or may not use the tool
response = agent("What is the capital of France?")  # Probably won't use tool
response = agent("Tell me about quantum entanglement")  # Will likely use tool

Force the model to always use a tool:

# pip install msgflux[openai] wikipedia
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

wikipedia = mf.Retriever.web_search("wikipedia")

class Researcher(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    tools = [wikipedia]
    config = {"tool_choice": "required", "verbose": True}

agent = Researcher()

# Model MUST call a tool before responding
response = agent("What is photosynthesis?")

Force a specific tool to be called:

# pip install msgflux[openai] wikipedia
import httpx
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

wikipedia = mf.Retriever.web_search("wikipedia")

def search_github(query: str) -> str:
    """Search GitHub repositories."""
    resp = httpx.get(
        "https://api.github.com/search/repositories",
        params={"q": query, "per_page": 5}
    )
    repos = resp.json().get("items", [])
    return "\n".join(f"- {r['full_name']}: {r['description']}" for r in repos)

class SearchAgent(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    tools = [search_github, wikipedia]
    # Always use GitHub
    config = {"tool_choice": "search_github", "verbose": True}

agent = SearchAgent()
response = agent("Find machine learning projects")

Disable tool usage temporarily:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

class Assistant(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    tools = [wikipedia, search_github]
    # Tools disabled
    config = {"tool_choice": "none", "verbose": True}

# Model will respond without using any tools
response = agent("What do you know about Python?")

Use required for routing agents:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

model = mf.Model.chat_completion("openai/gpt-4.1-mini")

@mf.tool_config(return_direct=True)
class PythonExpert(nn.Agent):
    """Expert in Python programming."""

    model = model
    system_message = "You are a Python expert."

@mf.tool_config(return_direct=True)
class RustExpert(nn.Agent):
    """Expert in Rust programming."""

    model = model
    system_message = "You are a Rust expert."

class Router(nn.Agent):
    """Routes programming questions to the right expert."""
    model = model
    system_message = "Route questions to the appropriate expert."
    tools = [PythonExpert, RustExpert]
    config = {"tool_choice": "required", "verbose": True}

router = Router()

# Router MUST pick an expert
response = router("How do I handle errors in Rust?")

Async Tools

When your agent runs asynchronously with acall(), prefer writing async tools as well. This ensures non-blocking execution and better performance when tools perform I/O operations.

Sync vs Async Tools

Async Tool (Recommended)Sync Tool

import httpx

async def fetch_data(url: str) -> str:
    """Fetch data from a URL asynchronously."""
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text

import httpx

def fetch_data(url: str) -> str:
    """Fetch data from a URL."""
    response = httpx.get(url, follow_redirects=True)
    return response.text

You can also implement a class-based async tool using the acall method:

Example

import httpx

class WebFetcher:
    """Fetch content from web pages."""

    def __init__(self, timeout: int = 30):
        self.timeout = timeout

    async def acall(self, url: str) -> str:
        """Fetch content from URL asynchronously."""
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.get(url)
            return response.text

Class-based Tools

Tools can be implemented as classes with __call__ or acall methods. This is useful when you need to maintain state or configure the tool at initialization.

Class-based Tool

Basic Class ToolOverride Tool Name

from typing import Optional
import httpx

class GitHubSearch:
    """Search GitHub repositories."""

    def __init__(self, max_results: Optional[int] = 5):
        self.max_results = max_results

    def __call__(self, query: str) -> str:
        """Search for repositories matching query.

        Args:
            query: Search term for repositories.
        """
        url = "https://api.github.com/search/repositories"
        params = {"q": query, "per_page": self.max_results}
        response = httpx.get(url, params=params, timeout=10)

        if response.status_code != 200:
            return f"Error: {response.status_code}"

        data = response.json()
        results = []
        for repo in data.get("items", []):
            results.append(f"- {repo['full_name']} ({repo['stargazers_count']}⭐)")

        return "\n".join(results) if results else "No repositories found."

Use the name attribute to override the class name:

import httpx

class GitHubRepoSearchV2:
    name = "search_repos"  # Exposed as "search_repos" instead of class name

    def __init__(self, max_results: int = 5):
        self.max_results = max_results

    def __call__(self, query: str) -> str:
        """Search GitHub for repositories."""
        url = "https://api.github.com/search/repositories"
        resp = httpx.get(url, params={"q": query, "per_page": self.max_results})
        repos = resp.json().get("items", [])
        return "\n".join(f"- {r['full_name']}" for r in repos) or "No results."

Return Types

Tools can return any data type. Non-string returns are automatically serialized using msgspec.json.encode before being passed to the model.

Tool Return Examples

String ReturnDict ReturnList Return

def add(a: float, b: float) -> str:
    """Sum two numbers."""
    return f"The sum of {a} plus {b} is {a + b}"

from typing import Dict

def web_search(query: str) -> Dict[str, str]:
    """Search for content."""
    return {
        "title": "Result title",
        "snippet": "Result snippet",
        "url": "https://example.com"
    }

from typing import List

def get_top_results(query: str) -> List[Dict]:
    """Get top search results."""
    return [
        {"title": "Result 1", "url": "..."},
        {"title": "Result 2", "url": "..."}
    ]

Tool Config

The @mf.tool_config decorator adds special behaviors to tools.

return_direct

When return_direct=True, the tool result is returned directly as the final response instead of going back to the model.

Use cases:

• Reduce agent calls by designing tools that return user-ready outputs
• Agent as router - delegate to specialists and return their responses directly

Example

Basic UsageWith Reasoning ModelsReport Generator

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

@mf.tool_config(return_direct=True)
def get_report() -> str:
    """Return the report."""
    return "This is your detailed report..."

class Assistant(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    tools = [get_report]

agent = Assistant()
response = agent("Give me the report")
# Returns the tool result directly, no model formatting

Combine return_direct with reasoning models to optimize tool calls. The model reasons about which tool to use, but the result bypasses additional processing:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(GROQ_API_KEY="...")

model = mf.Model.chat_completion(
    "groq/openai/gpt-oss-20b", reasoning_effort="low"
)

@mf.tool_config(return_direct=True)
def get_report() -> str:
    """Return the report from user."""
    return "This is your detailed report..."

class ReporterAgent(nn.Agent):
    model = model
    tools = [get_report]
    config = {"tool_choice": "required", "verbose": True}

agent = ReporterAgent()
response = agent("Give me the report")

Combine with inject_vars for external processing:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(GROQ_API_KEY="...")

@mf.tool_config(return_direct=True, inject_vars=True)
def generate_formatted_report(**kwargs) -> str:
    """Generate a formatted sales report."""
    vars = kwargs.get("vars", {})
    date_range = vars.get("date_range", "Unknown")

    # Mock data - in production, query your database
    report = f"""
    Sales Report: {date_range}
    ─────────────────────────────
    Total Revenue: $124,500
    Total Orders: 847
    Average Order: $147.04
    Top Product: Widget Pro (234 units)
    ─────────────────────────────
    Generated automatically.
    """
    return report

class Reporter(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    tools = [generate_formatted_report]
    config = {"verbose": True}

agent = Reporter()
response = agent("Generate the Q3 report", vars={"date_range": "2024-Q3"})

inject_vars

With inject_vars=True, tools can access and modify the agent's variable dictionary.

Use cases:

• Pass external credentials (API keys, tokens)
• Share state between tools
• Extract information from tools without returning it to the model (e.g., store metadata, logs, or intermediate results in vars for later use)

Example

External CredentialsNamed ParametersMutable State

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

@mf.tool_config(inject_vars=True)
def save_to_s3(**kwargs) -> str:
    """Save file to S3."""
    vars = kwargs.get("vars")
    token = vars["aws_token"]
    # Use token for S3 upload
    return "File saved successfully"

class Assistant(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    tools = [save_to_s3]

agent = Assistant()
response = agent("Save my file", vars={"aws_token": "secret-123"})

Inject specific vars as named parameters:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

@mf.tool_config(inject_vars=["api_key", "user_id"])
def upload_file(**kwargs) -> str:
    """Upload user file."""
    api_key = kwargs["api_key"]
    user_id = kwargs["user_id"]
    return f"Uploaded for user {user_id}"

class Assistant(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    tools = [upload_file]

agent = Assistant()
response = agent("Upload my file", vars={"api_key": "...", "user_id": "123"})

Tools can modify vars for persistent state:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

@mf.tool_config(inject_vars=True)
def save_preference(name: str, value: str, **kwargs):
    """Save a user preference."""
    vars = kwargs.get("vars")
    vars[name] = value  # Modifies the vars dict
    return f"Saved {name} = {value}"

@mf.tool_config(inject_vars=True)
def get_preference(name: str, **kwargs):
    """Get a user preference."""
    vars = kwargs.get("vars")
    return vars.get(name, "Not found")

class Assistant(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    tools = [save_preference, get_preference]

agent = Assistant()

user_vars = {}
agent("Save my favorite color as blue", vars=user_vars)
agent("What is my favorite color?", vars=user_vars)

print(user_vars)  # {"favorite_color": "blue"}

disable_input

With disable_input=True, the tool exposes no public input parameters to the model. The tool is called as tool_name(), and any arguments supplied by the model are ignored at runtime.

This is useful for:

• Specialist subagents that should be triggered without a task payload
• Tools that work only with injected context such as message, messages, or vars
• Internal routing tools where the coordinator should only decide whether to call

Example

import msgflux as mf
import msgflux.nn as nn

@mf.tool_config(disable_input=True, inject_messages=True)
class Specialist(nn.Agent):
    """Specialist that works only from conversation context."""

    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    system_message = "You are a specialist. Use the conversation history."

class Coordinator(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    tools = [Specialist]

inject_message

With inject_message=True, the tool receives the original message passed to the agent. This is useful when the tool needs access to the full envelope object, including Message fields that should not be part of the public tool schema.

Use cases:

• Agent-as-a-tool with declarative Message envelopes
• Tools that need response_mode side effects on the original message
• Access to vars, metadata, or other fields outside the public task schema

Example

@mf.tool_config(inject_message=True)
def inspect_original_message(message, **kwargs) -> str:
    """Inspect the original message envelope."""
    if isinstance(message, mf.Message):
        return str(message.get("meta.trace_id"))
    return "No structured message available."

inject_messages

With inject_messages=True, the tool receives the agent's internal state (conversation history) as messages in kwargs. This is particularly useful for agent-as-a-tool patterns where you want to pass the full conversation context to a specialist agent.

Use cases:

• Agent-as-a-tool: Pass conversation history to specialist agents
• Safety/moderation checks on conversation
• Access multimodal context (e.g. images in conversation)
• Context-aware tool execution

Example

Agent-as-Tool (Primary Use)Safety CheckerContext-Aware Processing

When an agent is used as a tool, inject_messages passes the conversation history so the specialist has full context:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

model = mf.Model.chat_completion("openai/gpt-4.1-mini")

# With inject_messages, the specialist receives
# the coordinator's conversation as messages
@mf.tool_config(inject_messages=True)
class Specialist(nn.Agent):
    """Expert that needs conversation context."""

    model = model
    system_message = "You are a specialist."

class Coordinator(nn.Agent):
    model = model
    system_message = "Route to specialists when needed."
    tools = [Specialist]
    config = {"verbose": True}

coordinator = Coordinator()

# When coordinator calls specialist, the full conversation
# is passed via messages parameter
response = coordinator("Help me with a complex problem")

Check conversation safety before responding:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

@mf.tool_config(inject_messages=True)
def check_safety(**kwargs) -> dict:
    """Check if the conversation is safe to continue."""
    messages = kwargs.get("messages", [])
    last_message = messages[-1]["content"] if messages else ""

    # Simple keyword-based safety check
    forbidden_keywords = ["hack", "exploit", "malware", "attack"]
    content_lower = last_message.lower()
    is_safe = not any(kw in content_lower for kw in forbidden_keywords)

    return {
        "safe": is_safe,
        "reason": None if is_safe else "Potentially harmful content detected"
    }

class SafeAgent(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    instructions = "Always check safety before responding."
    tools = [check_safety]
    config = {"verbose": True}

agent = SafeAgent()
response = agent("Can you help me write a Python script?")

Access images or other multimodal content from conversation:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

@mf.tool_config(inject_messages=True)
def analyze_shared_images(**kwargs) -> str:
    """Analyze all images shared in the conversation."""
    messages = kwargs.get("messages", [])

    images = []
    for msg in messages:
        content = msg.get("content", [])
        if isinstance(content, list):
            for block in content:
                if block.get("type") == "image_url":
                    images.append(block["image_url"]["url"])

    if not images:
        return "No images found in conversation."

    return f"Found {len(images)} images to analyze."

handoff

When handoff=True, the tool is configured for seamless agent-to-agent handoff:

• Sets return_direct=True and inject_messages=True
• Changes tool name to transfer_to_{original_name}
• Removes input parameters (equivalent to disable_input=True)

Unlike Agent-as-Tool, the Specialist's response bypasses the Coordinator entirely and goes directly to the user. The Coordinator only decides who handles the request.

              Input
                │
                ▼
  ┌────────────────────────────────┐
  │         Coordinator            │
  │                                │
  │   ┌──────────┐                 │
  │   │  Model   │──▶ "transfer_to │
  │   └──────────┘    Specialist()"│
  └──────────────┬─────────────────┘
                 │
                 │  + full conversation history
                 ▼
  ┌──────────────────────────────┐
  │         Specialist           │
  │          (Agent)             │
  │                              │
  │  receives full conversation  │
  │  context and takes ownership │
  └─────────────────┬────────────┘
                    │
                    ▼  (direct — Coordinator bypassed)
                  Output

Example

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

model = mf.Model.chat_completion("openai/gpt-4.1-mini")

# Tool is now "transfer_to_TechnicalSupport" with no parameters
@mf.tool_config(handoff=True)
class TechnicalSupport(nn.Agent):
    """Specialist for technical issues, debugging, and troubleshooting."""
    model = model
    system_message = "You are a technical support specialist."
    instructions = "Help users solve technical problems step by step."
    config = {"verbose": True}

class Coordinator(nn.Agent):
    """Routes user queries to the appropriate specialist."""
    model = model
    system_message = "You are a support coordinator."
    instructions = "Transfer users to technical support for technical issues."
    tools = [TechnicalSupport]
    config = {"verbose": True}

coordinator = Coordinator()
response = coordinator("My application crashes when I try to connect to the database")

call_as_response

Return tool call parameters without executing the tool. Useful for extracting structured data.

Use cases:

• BI report parameter extraction
• API call preparation
• Form data collection

Example

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

model = mf.Model.chat_completion("openai/gpt-4.1-mini")

@mf.tool_config(call_as_response=True)
def generate_sales_report(
    start_date: str, end_date: str, metrics: list[str], group_by: str
) -> dict:
    """Generate a sales report within a given date range.

    Args:
        start_date: Start date in YYYY-MM-DD format.
        end_date: End date in YYYY-MM-DD format.
        metrics: List of metrics to include (e.g., ["revenue", "orders", "profit"]).
        group_by: Dimension to group data by (e.g., "region", "product", "sales_rep").

    Returns:
        A structured sales report as a dictionary.
    """
    return  # Never executed

class BIAnalyst(nn.Agent):
    model = model
    system_message = """You're a BI analyst. When a user requests sales reports,
    you should simply complete the generate_sales_report tool call,
    extracting the requested metrics, dates, and groupings."""
    tools = [generate_sales_report]
    config = {"verbose": True}

agent = BIAnalyst()
response = agent(
    "I need a report of sales between July 1st and August 31st, 2025, "
    "showing revenue and profit, grouped by region."
)
# Returns the tool call parameters without executing the function

spawn

Dispatch a tool without waiting for a result. The model receives a confirmation that the task was started, but no return value. Requires async tool.

Use cases:

• Spawned operations (emails, notifications)
• Tasks that don't need to return a result to the model

Example

# pip install msgflux[openai]
import asyncio
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

@mf.tool_config(spawn=True)
async def send_notification(user_id: str, message: str):
    """Send notification asynchronously. Will not generate a return."""
    # Simulate async operation (e.g., API call, email sending)
    await asyncio.sleep(2)
    print(f"Notification sent to {user_id}: {message}")

class Notifier(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    tools = [send_notification]
    config = {"verbose": True}

agent = Notifier()

# Agent returns immediately, notification is dispatched
response = agent("Notify user123 that their order shipped")

name_override

Assign a custom name to a tool:

import httpx

@mf.tool_config(name_override="search_repos")
def github_repository_search_v2_extended(query: str) -> str:
    """Search GitHub repositories."""
    url = "https://api.github.com/search/repositories"
    resp = httpx.get(url, params={"q": query, "per_page": 3})
    repos = resp.json().get("items", [])
    return "\n".join(f"- {r['full_name']}" for r in repos)

# Tool is exposed as "search_repos" instead of the long function name

retry

Control retry behavior per tool. Accepts a tenacity decorator, False to disable, or None (default) to use env-based retry.

By default, all tools have automatic retry enabled using environment variables (TOOL_STOP_AFTER_ATTEMPT, TOOL_STOP_AFTER_DELAY). Use this parameter to customize or disable retry for specific tools.

Example

Custom RetryDisable RetryDefault (env-based)

from tenacity import retry, stop_after_attempt, wait_exponential

@mf.tool_config(
    retry=retry(
        reraise=True,
        stop=stop_after_attempt(5),
        wait=wait_exponential(min=1, max=10),
    )
)
def call_external_api(query: str) -> str:
    """Call an unreliable external API."""
    import httpx
    resp = httpx.get("https://api.example.com/search", params={"q": query})
    resp.raise_for_status()
    return resp.json()["result"]

@mf.tool_config(retry=False)
def fast_lookup(key: str) -> str:
    """Fast local lookup that should not retry."""
    return cache[key]

# No retry parameter needed — uses env defaults
@mf.tool_config(return_direct=True)
def search(query: str) -> str:
    """Search with default retry behavior."""
    return do_search(query)

Agent-as-Tool

Agents can be used as tools for other agents, enabling hierarchical task delegation, also known as SubAgents. Using AutoParams makes this pattern especially clean: the class name becomes the tool name, and the docstring becomes the tool description.

The Coordinator calls the Specialist as any other tool. The result returns to the Coordinator's model, which synthesizes it into the final response.

              Input
                │
                ▼
  ┌──────────────────────────────┐
  │         Coordinator          │
  │                              │
  │   ┌──────────┐               │
  │   │  Model   │──▶ "call      │
  │   └──────────┘    Specialist │
  └─────────────┬────────────────┘
                │  call(task)
                ▼
  ┌──────────────────────────────┐
  │         Specialist           │
  │        (SubAgent)            │
  │                              │
  │  processes task independently│
  │  may call its own tools      │
  └─────────────┬────────────────┘
                │  result
                ▼
  ┌──────────────────────────────┐
  │         Coordinator          │
  │                              │
  │   ┌──────────┐               │
  │   │  Model   │──▶ synthesized│
  │   └──────────┘    response   │
  └─────────────┬────────────────┘
                │
                ▼
                Output

Agent-as-Tool Examples

Health TeamResearch TeamAgent RouterHandoff Pattern

A coordinator agent delegates to specialist agents:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

model = mf.Model.chat_completion("openai/gpt-4.1-mini")

class Nutritionist(nn.Agent):
    """Specialist in nutrition, diet planning, and healthy eating habits.
    Consult for meal plans, dietary recommendations, and nutritional advice."""

    model = model
    system_message = "You are a certified nutritionist."
    instructions = """Create clear and practical meal plans tailored to the user's goals.
    Be objective, technical, and structured."""

class FitnessTrainer(nn.Agent):
    """Specialist in fitness, exercise routines, and physical training.
    Consult for workout plans, training schedules, and exercise guidance."""

    model = model
    system_message = "You are a certified personal trainer."
    instructions = """Design workout routines based on the user's fitness level and goals.
    Focus on safety, progression, and sustainability."""

class HealthCoordinator(nn.Agent):
    """Coordinates health specialists to provide comprehensive wellness advice."""

    model = model
    system_message = "You coordinate a team of health specialists."
    instructions = "Delegate user requests to the appropriate specialist."
    tools = [Nutritionist, FitnessTrainer]
    config = {"verbose": True}

coordinator = HealthCoordinator()

response = coordinator("I want to lose 10kg and build muscle")

Multiple research specialists with a coordinator:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

model = mf.Model.chat_completion("openai/gpt-4.1-mini")

class AcademicResearcher(nn.Agent):
    """Expert in academic research with peer-reviewed sources.
    Use for scholarly inquiries and scientific topics."""

    model = model
    system_message = "You are an academic researcher."
    expected_output = "Provide academic-level analysis with citations."

class MarketResearcher(nn.Agent):
    """Expert in market research and competitive analysis.
    Use for business intelligence and market sizing."""

    model = model
    system_message = "You are a market research analyst."
    expected_output = "Provide actionable business insights."

class TechnicalResearcher(nn.Agent):
    """Expert in technical documentation and APIs.
    Use for programming questions and library comparisons."""

    model = model
    system_message = "You are a technical researcher."
    expected_output = "Provide technical details with code examples."

class ResearchCoordinator(nn.Agent):
    model = model
    system_message = "You coordinate research specialists."
    instructions = "Delegate to the appropriate researcher based on the query type."
    tools = [
        AcademicResearcher,
        MarketResearcher,
        TechnicalResearcher
    ]
    config = {"verbose": True}

coordinator = ResearchCoordinator()

response = coordinator("Compare FastAPI vs Flask for building REST APIs")

Route requests directly to specialists using return_direct:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

model = mf.Model.chat_completion("openai/gpt-4.1-mini")

@mf.tool_config(return_direct=True)
class PythonExpert(nn.Agent):
    """Expert in Python performance optimization."""

    model = model
    system_message = "You specialize in Python performance."

@mf.tool_config(return_direct=True)
class JavaScriptExpert(nn.Agent):
    """Expert in JavaScript and Node.js."""

    model = model
    system_message = "You specialize in JavaScript."

class Router(nn.Agent):
    model = model
    system_message = "Route programming questions to the right expert."
    tools = [PythonExpert, JavaScriptExpert]
    config = {"verbose": True}

router = Router()

# Response comes directly from the specialist
response = router("How do I optimize a Python loop?")

Seamless conversation handoff between agents:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

model = mf.Model.chat_completion("openai/gpt-4.1-mini")

# Enable handoff - transfers conversation history
@mf.tool_config(handoff=True)
class StartupSpecialist(nn.Agent):
    """Specialist in scaling digital startups.
    Use for growth strategies, metrics, and funding."""

    model = model
    system_message = "You are a startup scaling expert."

class BusinessConsultant(nn.Agent):
    model = model
    system_message = """You are a business consultant.
    If the context is a startup, transfer to the specialist."""
    tools = [StartupSpecialist]
    config = {"verbose": True}

consultant = BusinessConsultant()

# Conversation is handed off to specialist
response = consultant(
    "My SaaS has a CAC of $120 and LTV of $600. How do I scale?"
)

MCP

The Model Context Protocol (MCP) allows agents to connect to external tool servers. MCP servers expose tools that can be called by the agent, enabling integration with filesystems, databases, APIs, and other services.

Configure MCP servers using the mcp_servers attribute:

Example

Stdio TransportHTTP TransportWith Tool ConfigurationBuild Your Own (FastMCP)

Connect to an MCP server via standard I/O:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

class FileAgent(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    mcp_servers = [{
        "name": "filesystem",
        "transport": "stdio",
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
    }]
    config = {"verbose": True}

agent = FileAgent()
response = agent("List all files in the current directory")

Connect to an MCP server via HTTP:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

class APIAgent(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    mcp_servers = [{
        "name": "api",
        "transport": "http",
        "base_url": "http://localhost:8000",
        "headers": {"Authorization": "Bearer token"}
    }]

agent = APIAgent()

Apply tool_config options to MCP tools:

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

class ConfiguredAgent(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    mcp_servers = [{
        "name": "filesystem",
        "transport": "stdio",
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
        "include_tools": ["read_file", "write_file"],
        "tool_config": {
            "read_file": {"inject_vars": True}
        }
    }]

agent = ConfiguredAgent()

Build a Python MCP server with FastMCP and connect it to an Agent — no Node.js required.

1. Create the server (my_server.py):

# /// script
# requires-python = ">=3.10"
# dependencies = ["fastmcp"]
# ///
"""MCP server — launch with: uv run my_server.py"""
from fastmcp import FastMCP

mcp = FastMCP("my-server")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

@mcp.tool()
def get_weather(city: str) -> str:
    """Return the current weather for a city."""
    # replace with a real API call
    return f"It's sunny in {city}, 24°C"

if __name__ == "__main__":
    mcp.run()

The # /// script block is uv inline script metadata. uv run my_server.py installs fastmcp automatically in an isolated environment — no pip install or pyproject.toml changes needed.

2. Connect via Agent (main.py):

# pip install msgflux[openai]
import msgflux as mf
import msgflux.nn as nn

# mf.set_envs(OPENAI_API_KEY="...")

class MyAgent(nn.Agent):
    model = mf.Model.chat_completion("openai/gpt-4.1-mini")
    instructions = "You are a helpful assistant."
    mcp_servers = [{
        "name": "my",
        "transport": "stdio",
        "command": "uv",
        "args": ["run", "my_server.py"],
    }]

agent = MyAgent()
response = agent("What is 3 + 4? Also, what's the weather in São Paulo?")
print(response)
# The agent has access to my__add and my__get_weather.

Tool namespacing

Tools are prefixed with the server name: my__add, my__get_weather. Use include_tools / exclude_tools to control which tools are exposed.

Server Configuration Options:

Option	Description
name	Namespace for tools from this server
transport	"stdio" or "http"
command	Command to start the server (stdio only)
args	Command arguments (stdio only)
cwd	Working directory (stdio only)
env	Environment variables (stdio only)
base_url	Server URL (http only)
headers	Additional HTTP headers (http only)
auth	Authentication provider — BearerTokenAuth, APIKeyAuth, etc. (http only)
include_tools	Allowlist of tools to expose
exclude_tools	Blocklist of tools to hide
tool_config	Per-tool configuration options