How to Build an MCP Style Routed AI Agent System with Dynamic Tool Exposure Planning, Execution, and Context Injection

In this tutorial, we build a fully functional MCP-style routed agent system from scratch, combining tool discovery, intelligent routing, structured planning, and execution into a single cohesive workflow. We start by setting up a modular tool server that exposes capabilities such as web search, local retrieval, dataset loading, and Python execution, all defined through structured schemas. We then implement a hybrid router that uses both heuristics and LLM reasoning to dynamically decide which tools to expose for a given task, ensuring minimal yet effective capability exposure. As we progress, we design an agent that plans tool usage, executes calls safely, and synthesizes final answers by injecting context from tool outputs. By the end, we demonstrate multiple real-world tasks and show how MCP principles such as context injection, routing policies, and restricted tool access come together to create a scalable, interpretable, and efficient agent system.

Copy CodeCopiedUse a different Browser

import sys import subprocess import pkgutil

def ensure_packages(): required = [ ("openai", "openai>=1.40.0"), ("pandas", "pandas"), ("numpy", "numpy"), ("sklearn", "scikit-learn"), ("pydantic", "pydantic"), ("duckduckgo_search", "duckduckgo-search"), ("rich", "rich"), ] missing = [] for import_name, pip_name in required: if pkgutil.find_loader(import_name) is None: missing.append(pip_name) if missing: subprocess.check_call([sys.executable, "-m", "pip", "install", "-q"] + missing)

ensure_packages()

import os import io import re import json import math import time import textwrap import traceback import contextlib from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Callable, Tuple

import numpy as np import pandas as pd

from openai import OpenAI from pydantic import BaseModel, Field from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity from duckduckgo_search import DDGS from rich.console import Console from rich.panel import Panel from rich.table import Table from rich.json import JSON as RichJSON

console = Console()

try: from google.colab import userdata OPENAI_API_KEY = userdata.get("OPENAI_API_KEY") except Exception: OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")

if not OPENAI_API_KEY: import getpass OPENAI_API_KEY = getpass.getpass("Enter OPENAI_API_KEY: ").strip()

os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY client = OpenAI(api_key=OPENAI_API_KEY)

MODEL = os.environ.get("OPENAI_MODEL", "gpt-4.1-mini") MAX_TOOL_CALLS = 3 MAX_WEB_RESULTS = 5 TOP_K_RETRIEVAL = 3

We begin by checking and installing all required Python packages so the tutorial runs smoothly in a single environment. We then import the core libraries for data handling, retrieval, structured schemas, web search, and rich console display. We securely load the OpenAI API key, initialize the client, and define global settings for the model, tool calls, web results, and retrieval depth.

Copy CodeCopiedUse a different Browser

class ToolSpec(BaseModel): name: str description: str input_schema: Dict[str, Any] tags: List[str] = Field(default_factory=list)

class ToolCall(BaseModel): tool_name: str arguments: Dict[str, Any]

class RouteDecision(BaseModel): selected_tools: List[str] rationale: str policy_notes: List[str] = Field(default_factory=list)

class PlanOutput(BaseModel): requires_tools: bool tool_calls: List[ToolCall] = Field(default_factory=list) direct_answer_allowed: bool = False planner_note: str = ""

class ToolResult(BaseModel): tool_name: str ok: bool output: Any error: Optional[str] = None

LOCAL_DOCS = [ { "id": "doc_001", "title": "Model Context Protocol Basics", "text": "Model Context Protocol standardizes how models connect to tools, resources, and prompts. A client can discover available tools from a server and invoke them using structured arguments." }, { "id": "doc_002", "title": "Dynamic Capability Exposure", "text": "Dynamic capability exposure means an agent does not always see every tool. A router can expose only the most relevant tools for a task, improving safety, reducing distraction, and lowering tool selection entropy." }, { "id": "doc_003", "title": "Context Injection for Agents", "text": "Context injection is the process of enriching the model prompt with selected tool descriptions, tool outputs, retrieved documents, prior summaries, and policy hints before the model generates a response." }, { "id": "doc_004", "title": "Tool Discovery and MCP", "text": "In MCP style systems, tool discovery usually begins with a tools listing step. Each tool includes a name, description, and input schema so the client knows how and when to call it." }, { "id": "doc_005", "title": "Router Policies for Agents", "text": "Routing policies can combine heuristics, learned scorers, confidence estimates, and LLM reasoning. A router may use task keywords, domain tags, or explicit constraints to decide which tools to expose." }, { "id": "doc_006", "title": "Why Restrict Tool Access", "text": "Restricting tool access helps minimize accidental misuse, improves reasoning focus, reduces latency, and creates a more interpretable planning process. This is especially helpful in multi-tool agent systems." }, { "id": "doc_007", "title": "Dataset Loading for Rapid Analysis", "text": "A dataset loader tool can let an agent inspect tabular data quickly. It is useful for classification tasks, summary statistics, schema exploration, and downstream code execution." }, { "id": "doc_008", "title": "Python Sandboxes in Agent Systems", "text": "Many advanced agents rely on code execution sandboxes for calculations, simulation, plotting, and dataframe inspection. Safe code execution typically uses restricted globals and output capture." }, ]

class LocalRetriever: def init(self, docs: List[Dict[str, str]]): self.docs = docs self.vectorizer = TfidfVectorizer(stop_words="english") self.doc_matrix = self.vectorizer.fit_transform([d["text"] for d in docs])

def search(self, query: str, top_k: int = 3) -> List[Dict[str, Any]]: q_vec = self.vectorizer.transform([query]) sims = cosine_similarity(q_vec, self.doc_matrix)[0] idxs = np.argsort(-sims)[:top_k] results = [] for i in idxs: results.append({ "id": self.docs[i]["id"], "title": self.docs[i]["title"], "text": self.docs[i]["text"], "score": float(sims[i]), }) return results

retriever = LocalRetriever(LOCAL_DOCS)

We define structured Pydantic models to represent tool specifications, tool calls, routing decisions, planning outputs, and tool results in a clean MCP-style format. We then create a small local knowledge base that explains concepts like MCP, dynamic capability exposure, context injection, router policies, and sandboxed execution. Finally, we built a TF-IDF-based local retriever that searches these documents and returns the most relevant snippets, along with their similarity scores.

Copy CodeCopiedUse a different Browser

def tool_web_search(query: str, max_results: int = 5) -> Dict[str, Any]: results = [] with DDGS() as ddgs: for r in ddgs.text(query, max_results=max_results): results.append({ "title": r.get("title", ""), "href": r.get("href", ""), "body": r.get("body", ""), }) return {"query": query, "results": results}

def tool_python_exec(code: str) -> Dict[str, Any]: allowed_builtins = { "abs": abs, "all": all, "any": any, "bool": bool, "dict": dict, "enumerate": enumerate, "float": float, "int": int, "len": len, "list": list, "max": max, "min": min, "print": print, "range": range, "round": round, "set": set, "sorted": sorted, "str": str, "sum": sum, "tuple": tuple, "zip": zip, }

local_ns = {} global_ns = { "builtins": allowed_builtins, "np": np, "pd": pd, "math": math, }

stdout_buffer = io.StringIO() try: with contextlib.redirect_stdout(stdout_buffer): exec(code, global_ns, local_ns) return { "stdout": stdout_buffer.getvalue(), "locals": {k: repr(v)[:500] for k, v in local_ns.items() if not k.startswith("")} } except Exception as e: return { "stdout": stdout_buffer.getvalue(), "error_type": type(e).name__, "error_message": str(e), "traceback": traceback.format_exc(limit=2), }

def load_builtin_dataset(name: str = "iris", n_rows: int = 10) -> Dict[str, Any]: from sklearn import datasets as sk_datasets registry = { "iris": sk_datasets.load_iris, "wine": sk_datasets.load_wine, "breast_cancer": sk_datasets.load_breast_cancer, "diabetes": sk_datasets.load_diabetes, } if name not in registry: raise ValueError(f"Unsupported dataset '{name}'. Choose from {list(registry.keys())}") ds = registry[name]() feature_names = list(ds.feature_names) df = pd.DataFrame(ds.data, columns=feature_names) if hasattr(ds, "target"): df["target"] = ds.target return { "dataset_name": name, "shape": list(df.shape), "columns": list(df.columns), "preview": df.head(n_rows).to_dict(orient="records"), "describe": df.describe(include="all").fillna("").to_dict(), }

def tool_vector_retrieve(query: str, top_k: int = 3) -> Dict[str, Any]: results = retriever.search(query, top_k=top_k) return {"query": query, "results": results}

We define the main tools our MCP-style agent can use, including web search, safe Python execution, dataset loading, and local vector retrieval. We keep Python execution controlled by limiting built-in functions and capturing printed output, local variables, and errors. We also ensure that the dataset and retrieval tools return structured outputs so the agent can inspect data or retrieve relevant knowledge before generating a final answer.

Copy CodeCopiedUse a different Browser

@dataclass class MCPTool: spec: ToolSpec fn: Callable[..., Any]

class MCPToolServer: def init(self): self.tools: Dict[str, MCPTool] = {}

def register_tool(self, spec: ToolSpec, fn: Callable[..., Any]): self.tools[spec.name] = MCPTool(spec=spec, fn=fn)

def tools_list(self) -> List[Dict[str, Any]]: return [ { "name": tool.spec.name, "description": tool.spec.description, "input_schema": tool.spec.input_schema, "tags": tool.spec.tags, } for tool in self.tools.values() ]

def tools_call(self, tool_name: str, arguments: Dict[str, Any]) -> ToolResult: if tool_name not in self.tools: return ToolResult(tool_name=tool_name, ok=False, output=None, error="Tool not found") try: output = self.tools[tool_name].fn(**arguments) return ToolResult(tool_name=tool_name, ok=True, output=output) except Exception as e: return ToolResult(tool_name=tool_name, ok=False, output=None, error=f"{type(e).name}: {str(e)}")

server = MCPToolServer()

server.register_tool( ToolSpec( name="web_search", description="Search the public web for recent or general information and return concise results.", input_schema={ "type": "object", "properties": { "query": {"type": "string"}, "max_results": {"type": "integer", "default": 5} }, "required": ["query"] }, tags=["web", "search", "recent", "news", "research"] ), tool_web_search )

server.register_tool( ToolSpec( name="python_exec", description="Execute Python code for calculations, dataframe inspection, simulations, or transformations.", input_schema={ "type": "object", "properties": { "code": {"type": "string"} }, "required": ["code"] }, tags=["python", "compute", "analysis", "code", "math"] ), tool_python_exec )

server.register_tool( ToolSpec( name="vector_retrieve", description="Retrieve relevant local knowledge snippets from a vectorized tutorial corpus.", input_schema={ "type": "object", "properties": { "query": {"type": "string"}, "top_k": {"type": "integer", "default": 3} }, "required": ["query"] }, tags=["retrieval", "memory", "knowledge", "vector", "rag"] ), tool_vector_retrieve )

server.register_tool( ToolSpec( name="dataset_loader", description="Load a built-in tabular dataset and return schema, preview, and summary statistics.", input_schema={ "type": "object", "properties": { "name": {"type": "string", "enum": ["iris", "wine", "breast_cancer", "diabetes"]}, "n_rows": {"type": "integer", "default": 10} }, "required": ["name

[truncated for AI cost control]