	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Tool Discovery and Dynamic Tool Loading

MCP Tool Discovery and Dynamic Tool Loading

Author: Venkata Sudhakar

MCPToolset loads all tools from a server by default. But in large deployments an MCP server might expose dozens of tools, and an agent only needs a subset. Loading all tools increases the context window and can confuse the model into choosing the wrong tool. Dynamic tool loading lets you inspect the available tools after connection and filter down to exactly what the agent needs before constructing it.

MCPToolset.from_server() returns the full tool list. You can inspect each tool by name and description, then pass only the filtered subset to the agent. This is also useful for role-based access: a read-only analyst agent gets query tools, while a write-enabled operations agent gets mutation tools from the same MCP server.

The below example shows a ShopMax India MCP server with both read and write tools, and two ADK agents that each load only their permitted subset.

# shopmax_mixed_server.py - MCP server with read and write tools
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("shopmax-mixed")

ORDERS = [
    {"id": "ORD-001", "customer": "Priya", "product": "Laptop", "status": "Pending"},
    {"id": "ORD-002", "customer": "Arjun", "product": "TV", "status": "Shipped"},
]

# READ tools
@mcp.tool()
def list_orders() -> list:
    # List all orders (read-only)
    return ORDERS

@mcp.tool()
def get_order(order_id: str) -> dict:
    # Get a single order by ID (read-only)
    for o in ORDERS:
        if o["id"] == order_id:
            return o
    return {"error": "Order not found"}

# WRITE tools
@mcp.tool()
def update_order_status(order_id: str, new_status: str) -> dict:
    # Update order status (write operation)
    for o in ORDERS:
        if o["id"] == order_id:
            o["status"] = new_status
            return {"updated": True, "order_id": order_id, "status": new_status}
    return {"error": "Order not found"}

@mcp.tool()
def cancel_order(order_id: str, reason: str) -> dict:
    # Cancel an order (write operation)
    for o in ORDERS:
        if o["id"] == order_id:
            o["status"] = "Cancelled"
            return {"cancelled": True, "order_id": order_id, "reason": reason}
    return {"error": "Order not found"}

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

Now connect agents with dynamic tool filtering by role,

import asyncio
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types

READ_TOOLS = ["list_orders", "get_order"]
WRITE_TOOLS = ["update_order_status", "cancel_order"]

async def create_agent(role: str):
    all_tools, exit_stack = await MCPToolset.from_server(
        connection_params=StdioServerParameters(command="python", args=["shopmax_mixed_server.py"])
    )
    # Discover and filter tools by role
    allowed_names = READ_TOOLS if role == "analyst" else READ_TOOLS + WRITE_TOOLS
    filtered_tools = [t for t in all_tools if t.name in allowed_names]
    print(f"[{role}] Loaded tools: {[t.name for t in filtered_tools]}")

agent = LlmAgent(
        name=f"shopmax_{role}",
        model="gemini-2.0-flash",
        tools=filtered_tools,
        instruction=f"You are a ShopMax India {role}. Use only your assigned tools."
    )
    return agent, exit_stack

async def main():
    # Analyst - read only
    analyst, stack1 = await create_agent("analyst")
    # Operations - read and write
    ops, stack2 = await create_agent("operations")

session_service = InMemorySessionService()
    for agent, stack, user_id, query in [
        (analyst, stack1, "u1", "List all orders."),
        (ops, stack2, "u2", "Update ORD-001 status to Delivered."),
    ]:
        runner = Runner(agent=agent, app_name=agent.name, session_service=session_service)
        session = session_service.create_session(app_name=agent.name, user_id=user_id)
        msg = types.Content(role="user", parts=[types.Part(text=query)])
        async for event in runner.run_async(user_id=user_id, session_id=session.id, new_message=msg):
            if event.is_final_response():
                print(f"[{agent.name}] {event.content.parts[0].text}")
        await stack.aclose()

asyncio.run(main())

It gives the following output,

[analyst] Loaded tools: ['list_orders', 'get_order']
[operations] Loaded tools: ['list_orders', 'get_order', 'update_order_status', 'cancel_order']

[shopmax_analyst] Current orders:
1. ORD-001 - Priya, Laptop - Status: Pending
2. ORD-002 - Arjun, TV - Status: Shipped

[shopmax_operations] Order ORD-001 has been updated to Delivered successfully.

Tool discovery at runtime also lets you build adaptive agents that inspect what a new MCP server offers before deciding how to use it. Print the tool names and descriptions after connection during development to understand what the server exposes. In production, cache the tool list after the first discovery rather than re-fetching on every agent startup. Use consistent tool naming conventions across your MCP servers - prefix read tools with get_ or list_, and write tools with create_, update_, or delete_ - so role-based filtering remains simple and reliable.

Send your comments, suggestions or queries regarding this site to [email protected].