# Designing Tool Topologies for AI Agents

> Source: <https://pub.towardsai.net/designing-tool-topologies-for-ai-agents-d0bf2ff33842?source=rss----98111c9905da---4>
> Published: 2026-11-02 09:42:15+00:00

The first step in designing a tool is to define its parameters clearly. These definitions guide the large language model (LLM) in identifying the information it needs to extract from the user’s query to call the tool. Clear guidance on interpreting the tool’s response also helps the LLM extract relevant information and provide an accurate answer to the user.

Once a tool’s parameters are defined, the LLM can request its execution. The tool can run locally or remotely, for example through an API call or a database query. The agent should handle latency and timeouts gracefully when calling tools through APIs. Timeout values may need to be adjusted to accommodate the tool’s expected response time.

Consider a traditional money-transfer application. When a user requests a transfer, the system follows a fixed sequence defined in the application code, with the appropriate dependencies and error handling. It might begin by checking whether the requested amount exceeds the available balance. If it does, the system declines the request; otherwise, it proceeds to validate the beneficiary, run fraud and compliance checks, debit the sender’s account, credit the recipient’s account, and send a confirmation. The developer defines the permitted paths, including how failures are handled, ahead of time.

An AI agent can approach the same request differently. Rather than following only a hard-coded sequence, the agent may be given a goal, such as “transfer ₹5,000 to Priya,” and tools such as get_balance, validate_beneficiary, check_fraud_risk, debit_account, credit_account, and send_notification. Within the workflow’s constraints, the model selects which tool to call, in what order, and how to respond to each result. For example, it might check the balance before validating the beneficiary. If a fraud check returns a warning, the workflow should pause or escalate according to policy; user confirmation alone should not override required safeguards. Some sequencing decisions can therefore be made at runtime, while critical controls remain enforced by the application.

This is where **tool topology** becomes important. Here, tool topology refers to how tools are arranged, connected, and exposed to the agent: which tools are available, which depend on others, which can run in parallel, and which require prior conditions to be met. In a traditional application, these relationships are often embedded in the code. In an agentic system, they should be explicitly defined and enforced. Tool descriptions alone do not prevent an agent from requesting a debit before a balance check or retrying a credit after a timeout. Backend validation, transaction controls, and duplicate-request protection are needed to prevent unsafe outcomes.

The design question therefore expands. Alongside asking “What is the correct sequence of steps?” we ask “How should tools and their controls be designed so that the agent can act flexibly without bypassing required safeguards?” Several approaches can help:

· **Enforce dependencies in the tools.** For example, debit_account can require a validated authorization token. The backend must still verify the current balance and required checks when executing the transaction.

· **Group tools into stages.** Expose payment tools only after the required validation steps have succeeded, and enforce the same prerequisites in the backend.

· **Combine critical steps into one tool.** A single execute_transfer tool can keep the debit and credit together when backed by an atomic transaction and duplicate-request protection. The agent can still handle flexible tasks, such as clarifying an ambiguous beneficiary name.

Building reliable AI agents requires choosing a tool topology that balances flexibility with control. The model needs enough freedom to handle unexpected situations, while the application enforces the deterministic safeguards required for high-stakes actions such as moving money.

Let’s explore a range of tool topologies and their trade-offs.

Let’s start with a task that requires a single tool call. The agent’s job is to select the most appropriate tool and invoke it once. The flow has five steps:

For example, a customer asks the agent for their account balance. Assuming the customer is authenticated and the account has been identified, answering requires a single tool call: the agent selects get_balance, passes the authorized account ID, receives the balance, and replies with the amount.

Single-tool execution is the simplest topology and provides a foundation for more complex architectures.

Sometimes a request requires several actions, and running them one after another would slow the response. In these cases, the agent can execute multiple tools in parallel. This raises a key question: how does the agent decide which tools to run, and how many?

One approach is semantic tool selection. The agent compares the meaning of the user’s query with the descriptions of available tools to identify relevant candidates. It then selects the tools needed to answer the request, subject to permissions, dependencies, and execution limits. The flow is:

3. **Call the tools in parallel.** Schedule independent tool calls concurrently, subject to concurrency limits.

**Illustrative example:** A customer asks, *“Give me a quick overview of my account.”* In this example, the overview uses three independent pieces of information, so the agent selects three tools and runs them together:

None of these tools depends on another’s output, so there is no reason to wait for one to finish before starting the next. If each call takes about one second, running them in sequence takes three seconds, while running them in parallel takes roughly one. The agent then combines the results into a single reply:

*“Your balance is ₹42,500. Your last five transactions include a ₹2,300 grocery payment and a ₹15,000 salary credit. You have an electricity bill of ₹1,800 due this Friday.”*

One important rule applies: calls in a parallel batch must be safe to run concurrently. If one tool requires another’s output or successful completion, those calls must run in sequence. For example, a transfer must satisfy the required balance and authorization checks before execution.

Running this prints the combined overview after about one second rather than three. Three details in the code are worth calling out:

· **asyncio.gather** schedules the selected asynchronous tool calls concurrently and collects their results. It does not make blocking, synchronous operations concurrent on its own.

· **asyncio.wait_for(…, timeout=3.0)** sets a timeout for the awaited operation. Handle timeout errors explicitly; cancellation and cleanup can extend the elapsed time beyond the configured timeout.

Sample Code:

``` python
import asyncioimport math# --- Tools: each is independent, so they can run concurrently ---async def get_balance(account_id: str) -> dict:    await asyncio.sleep(1)  # simulate a network call    return {"balance": 42500}async def get_transactions(account_id: str, limit: int = 5) -> dict:    await asyncio.sleep(1)    return {"transactions": [        {"desc": "Grocery store", "amount": -2300},        {"desc": "Salary credit", "amount": 15000},    ][:limit]}async def get_payments(account_id: str, days: int = 7) -> dict:    await asyncio.sleep(1)    return {"payments": [{"desc": "Electricity bill", "amount": 1800, "due": "Friday"}]}TOOLS = {    "get_balance":      {"fn": get_balance,      "desc": "current account balance"},    "get_transactions": {"fn": get_transactions, "desc": "recent account transactions history"},    "get_payments":     {"fn": get_payments,     "desc": "upcoming bills and scheduled payments"},}# --- Semantic tool selection (toy embedding for illustration) ---def embed(text: str) -> dict:    # Stand-in for a real embedding model: a simple bag-of-words vector.    vec = {}    for word in text.lower().split():        vec[word] = vec.get(word, 0) + 1    return vecdef cosine(a: dict, b: dict) -> float:    dot = sum(a[w] * b.get(w, 0) for w in a)    norm = math.sqrt(sum(v * v for v in a.values())) * math.sqrt(sum(v * v for v in b.values()))    return dot / norm if norm else 0.0def select_tools(query: str, threshold: float = 0.1) -> list[str]:    q = embed(query)    return [name for name, t in TOOLS.items() if cosine(q, embed(t["desc"])) >= threshold]# --- Parallel execution ---def build_params(tool_name: str, account_id: str) -> dict:    params = {"account_id": account_id}    if tool_name == "get_transactions":        params["limit"] = 5    if tool_name == "get_payments":        params["days"] = 7    return paramsasync def run_parallel(query: str, account_id: str, timeout: float = 3.0) -> dict:    selected = select_tools(query)    calls = [TOOLS[name]["fn"](**build_params(name, account_id)) for name in selected]    # gather runs all calls concurrently; return_exceptions keeps one failure    # from cancelling the others, so we can still return partial results.    results = await asyncio.wait_for(        asyncio.gather(*calls, return_exceptions=True), timeout=timeout    )    return dict(zip(selected, results))def combine(results: dict) -> str:    parts = []    if isinstance(results.get("get_balance"), dict):        parts.append(f"Your balance is ₹{results['get_balance']['balance']:,}.")    if isinstance(results.get("get_transactions"), dict):        recent = ", ".join(t["desc"] for t in results["get_transactions"]["transactions"])        parts.append(f"Recent activity: {recent}.")    if isinstance(results.get("get_payments"), dict):        for p in results["get_payments"]["payments"]:            parts.append(f"{p['desc']} of ₹{p['amount']:,} is due {p['due']}.")    failed = [name for name, r in results.items() if isinstance(r, Exception)]    if failed:        parts.append(f"(Couldn't fetch: {', '.join(failed)}.)")    return " ".join(parts)async def main():    query = "show my balance, recent transactions and upcoming payments"    results = await run_parallel(query, account_id="acct_123")    print(combine(results))asyncio.run(main())
```

**Things to watch for in production:**

· **Hidden dependencies.** Read-only, independent tools are a good fit for parallel batches. State-changing operations require additional safeguards for ordering, consistency, and duplicate requests; financial transfers should use a controlled transactional workflow.

A chain links steps together in sequence. Chains are useful when a user’s request requires multiple tools and the output or successful completion of one step is needed before the next can run. Execution proceeds until the task is complete or an unhandled failure stops the chain.

The money transfer from the introduction is a natural example. Required balance, beneficiary, and authorization checks must succeed before the transfer is executed. The debit and credit should be handled together by the transactional transfer service, rather than exposed as independently retried steps.

The most important part of building a chain is its design: the order of steps, the data passed between them, and the handling of failures. A failed prerequisite must prevent dependent actions from running. Error handling may still perform cleanup, report the failure, or escalate the request.

**Building chains with LCEL.** LangChain simplifies this with the LangChain Expression Language (LCEL), which lets you build chains by composing existing Runnables with the | (pipe) operator instead of manually wiring up Chain objects. Each step receives the output of the previous step, so dependencies are expressed by the order of the pipe itself.

**Example: money transfer chain**

A customer says, *“Send ₹5,000 to Priya.”* The chain runs five dependent steps:

4. **Execute the transfer.** After all required authorization, fraud, and compliance checks succeed, debit the sender and credit the recipient through the transactional transfer service.

A few things to notice:

· **The risky part requires atomicity.** Combining debit and credit in one tool is not enough by itself. The underlying service must enforce an atomic transaction and duplicate-request protection so retries cannot move money twice.

``` python
from pydantic import BaseModelfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnableLambdafrom langchain.chat_models import init_chat_model# --- Step 1: LLM extracts structured parameters from the user's message ---class TransferRequest(BaseModel):    amount: float    beneficiary: strllm = init_chat_model("your-model-name")  # any supported chat modelprompt = ChatPromptTemplate.from_template(    "Extract the transfer amount and beneficiary name from: {message}")extract = prompt | llm.with_structured_output(TransferRequest)# --- Steps 2-5: tools, each depending on the previous step's output ---ACCOUNT = {"id": "acct_123", "balance": 42500}PAYEES = {"priya": "acct_987"}class TransferError(Exception):    passdef check_balance(req: TransferRequest) -> dict:    if req.amount > ACCOUNT["balance"]:        raise TransferError(f"Insufficient balance: ₹{ACCOUNT['balance']:,} available.")    return {"amount": req.amount, "beneficiary": req.beneficiary}def validate_beneficiary(state: dict) -> dict:    payee_account = PAYEES.get(state["beneficiary"].lower())    if not payee_account:        raise TransferError(f"{state['beneficiary']} is not a registered payee.")    return {**state, "to_account": payee_account}def execute_transfer(state: dict) -> dict:    # Debit and credit happen together in one atomic operation,    # so the system never ends up with a debit but no credit.    ACCOUNT["balance"] -= state["amount"]    return {**state, "reference": "TXN-20261031-0042"}def confirm(state: dict) -> str:    return (f"₹{state['amount']:,.0f} sent to {state['beneficiary']}. "            f"Reference: {state['reference']}. "            f"New balance: ₹{ACCOUNT['balance']:,}.")# --- Compose the chain with LCEL: order of the pipe = order of dependencies ---transfer_chain = (    extract    | RunnableLambda(check_balance)    | RunnableLambda(validate_beneficiary)    | RunnableLambda(execute_transfer)    | RunnableLambda(confirm))try:    print(transfer_chain.invoke({"message": "Send ₹5,000 to Priya"}))except TransferError as e:    print(f"Transfer declined: {e}")
```

LCEL also supports production features on any step, such as .with_retry() for transient failures and .with_fallbacks() for a backup path. For flows that need branching, loops, or pausing for human approval (for example, confirming large transfers), LangChain points to LangGraph, which it describes as a framework to build agents with low-level control. [langchain](https://www.langchain.com/resources/langchain-alternatives)

**Alternatives to LangChain.** LangChain isn’t the only option. LangChain’s own comparison groups the alternatives into two layers: framework alternatives, where you want a different Python or TypeScript library for building agents, and platform alternatives, where you want a different runtime, observability and evaluation stack, or deployment layer. The framework-level options are the relevant ones for building chains: [langchain](https://www.langchain.com/resources/langchain-alternatives)

*Source:* *LangChain Alternatives: A Full Comparison for Agent Engineering**, LangChain, June 2026.*

The same page suggests when another tool may fit better: LlamaIndex if your use case is retrieval-heavy RAG without multi-step reasoning, CrewAI if fast multi-agent prototyping is the only need, Microsoft Agent Framework if you’re all-in on Azure, and Temporal if durable execution of long-running workflows is the main need and LLM steps are a bolt-on. [langchain](https://www.langchain.com/resources/langchain-alternatives)

What if the agent needs a workflow with multiple decision points? A simple linear chain is not the clearest representation for complex branching, loops, or human checkpoints. A graph models these transitions explicitly; LangGraph is one framework for building such workflows.

In a graph:

Graphs are incredibly flexible for modeling complex workflows. They let you **branch** (take different paths based on a condition), **run steps in parallel** (fan out to several nodes at once), **merge** (wait for parallel branches and combine their results), and **loop** (return to an earlier step, for example to retry). This lets you consolidate many tool executions into one unified process.

That flexibility comes with trade-offs. Graphs can be harder to design, test, and debug because they allow more possible execution paths. Cost and latency depend on the implementation, especially the number of model calls and external operations, rather than on the graph structure alone. Use code-based rules for deterministic decisions, such as comparing an amount with a balance, and reserve LLM calls for steps that require interpretation or judgment.

**Example: risk-aware money transfer**

Let’s extend the money-transfer example from the Chains section with an illustrative risk policy. A ₹500 transfer to a known payee may proceed after all required checks, while a ₹75,000 transfer may require human review and a new payee may require additional verification. These amounts illustrate the workflow; actual thresholds and controls depend on the institution’s policy. The graph brings these paths together:

1. **Extract parameters.** The LLM extracts the amount and beneficiary from the customer’s message.

4. **Route by risk (merge).** Wait for both checks, then enforce all applicable controls. A transfer may require both human approval and new-payee verification:

o **High value:** pause for **human_approval**. Approval allows the workflow to continue through any remaining checks; rejection leads to *Decline*.

o **New payee:** require one-time-password verification through **verify_otp**. Continue only after successful verification.

o **Low risk:** proceed to the transfer only after all required checks have succeeded.

5. **execute_transfer.** All authorized paths converge here. The backend executes debit and credit as an atomic transaction with duplicate-request protection.

6. **notify.** The customer gets a confirmation.

**Sample code (LangGraph)**

The tools below are simplified stand-ins. In a real system, extract would be an LLM call with structured output (as in the chains example), and the other nodes would call your banking services.

``` python
from typing import TypedDictfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.checkpoint.memory import MemorySaverfrom langgraph.types import interrupt, Command# --- Shared state: every node reads from it and adds to it ---class TransferState(TypedDict, total=False):    message: str    amount: float    beneficiary: str    balance: float    risk: str    payee_known: bool    approved: bool    reference: str    status: strACCOUNT_BALANCE = 150000KNOWN_PAYEES = {"priya", "rahul"}HIGH_VALUE_LIMIT = 50000# --- Nodes ---def extract(state: TransferState) -> dict:    # In practice: an LLM call with structured output.    return {"amount": 75000, "beneficiary": "Priya"}def check_balance(state: TransferState) -> dict:    return {"balance": ACCOUNT_BALANCE}def fraud_score(state: TransferState) -> dict:    return {"risk": "high" if state["amount"] >= HIGH_VALUE_LIMIT else "low"}def check_payee(state: TransferState) -> dict:    return {"payee_known": state["beneficiary"].lower() in KNOWN_PAYEES}def route(state: TransferState) -> dict:    return {}  # merge point: runs only after both parallel checks finishdef human_approval(state: TransferState) -> dict:    # Pauses the graph until a human responds, then resumes from here.    decision = interrupt({        "question": f"Approve ₹{state['amount']:,.0f} to {state['beneficiary']}?"    })    return {"approved": decision == "approve"}def verify_otp(state: TransferState) -> dict:    # In practice: send an OTP and wait for the customer to enter it.    return {}def execute_transfer(state: TransferState) -> dict:    # Debit and credit together, atomically.    return {"reference": "TXN-20261031-0077"}def notify(state: TransferState) -> dict:    return {"status": f"₹{state['amount']:,.0f} sent to {state['beneficiary']}. "                      f"Reference: {state['reference']}."}def decline(state: TransferState) -> dict:    return {"status": "Transfer declined."}# --- Decision functions (the conditional edges) ---def after_balance(state: TransferState):    if state["amount"] > state["balance"]:        return "decline"    return ["fraud_score", "check_payee"]  # returning a list fans out in paralleldef route_by_risk(state: TransferState) -> str:    if state["risk"] == "high":        return "human_approval"    if not state["payee_known"]:        return "verify_otp"    return "execute_transfer"def after_approval(state: TransferState) -> str:    return "execute_transfer" if state["approved"] else "decline"# --- Build the graph ---builder = StateGraph(TransferState)for name, fn in [    ("extract", extract), ("check_balance", check_balance),    ("fraud_score", fraud_score), ("check_payee", check_payee),    ("route", route), ("human_approval", human_approval),    ("verify_otp", verify_otp), ("execute_transfer", execute_transfer),    ("notify", notify), ("decline", decline),]:    builder.add_node(name, fn)builder.add_edge(START, "extract")builder.add_edge("extract", "check_balance")builder.add_conditional_edges("check_balance", after_balance,                              ["decline", "fraud_score", "check_payee"])builder.add_edge(["fraud_score", "check_payee"], "route")  # wait for bothbuilder.add_conditional_edges("route", route_by_risk,                              ["human_approval", "verify_otp", "execute_transfer"])builder.add_conditional_edges("human_approval", after_approval,                              ["execute_transfer", "decline"])builder.add_edge("verify_otp", "execute_transfer")builder.add_edge("execute_transfer", "notify")builder.add_edge("notify", END)builder.add_edge("decline", END)# A checkpointer saves state so the graph can pause and resume.graph = builder.compile(checkpointer=MemorySaver())# --- Run it ---config = {"configurable": {"thread_id": "txn-001"}}result = graph.invoke({"message": "Send ₹75,000 to Priya"}, config)if "__interrupt__" in result:    print(result["__interrupt__"][0].value["question"])    # A reviewer approves; the graph resumes exactly where it paused.    result = graph.invoke(Command(resume="approve"), config)print(result["status"])
```

What to notice in the code:

Under the illustrative policy, a ₹75,000 transfer follows the high-value review path and completes only if approval and all other required checks succeed. A ₹500 transfer to a known payee can take the low-risk path. A ₹2,00,000 request is declined if it exceeds the available balance.

**When to use graphs.** Choose a graph when the workflow needs branching, merging, loops, or human checkpoints. A chain is often simpler when steps always run in the same order. Independent, concurrency-safe calls may need only parallel execution. Use the simplest topology that satisfies the workflow’s requirements.

**Further reading:** To learn about different tool-selection strategies, refer to “How AI Agents Choose the Right Tool” on Medium. Article reference

https://medium.com/@dkum062112_8007/how-ai-agents-choose-the-right-tool-5c07344a1a9c?sharedUserId=dkum062112_8007

[Designing Tool Topologies for AI Agents](https://pub.towardsai.net/designing-tool-topologies-for-ai-agents-d0bf2ff33842) was originally published in [Towards AI](https://pub.towardsai.net) on Medium, where people are continuing the conversation by highlighting and responding to this story.
