Ideal PodiumBuilding an MCP Client and Server from Scratch with an Open LLM
Vikas Kumar Ojha
MachineLearning@Avalara Ex-Samsung
If you're a sci-fi fan, you've probably wished at some point for a real-life Jarvis. While true AGI is still far off and what “AGI” even means is a debate for another day, deep learning has advanced dramatically, shifting from narrow, domain-specific systems to highly generalized models. Modern chatbots built with generative techniques feel remarkably human. Even more fascinating is the rise of the MCP (Model Context Protocol), which enables LLMs to take meaningful actions in the real world, making them exponentially more useful. ChatGPT was already impressive with its coding and writing abilities, but when paired with MCP-enabled applications like Cursor, it can genuinely feel like a capable teammate, provided it is used with care.
Most tutorials I've come across focus on building an MCP server and then rely on existing client orchestrators such as Claude Desktop, Cursor, or LLM Studio. Instead, I wanted to take a different approach: building the client orchestration layer from scratch. This not only gives me far more flexibility but also forces me to deeply understand and re-implement the orchestration mechanics that power MCP clients.
Understanding Model Context Protocol
Before we dive into implementation, it's better to understand how MCP works. Let's begin with the most fundamental question.
What is the Model Context Protocol (MCP)?
To simplify, MCP is a standard that a MCP client would use to interact with the MCP servers.
Before we dive into how it all works, let's first break down the core components involved. Once we understand these building blocks, it will become much easier to see how they come together to accomplish the task at hand.
MCP Server
An MCP server is a standalone component that receives requests from a client, typically in JSON or another supported format, performs an action, and returns a processed response.
I don't want to make this sound overly technical, so I put together a simple diagram to show the main idea. Basically, an MCP server is just a service that talks to your app or hits internal/external APIs, does whatever work you ask it to do, and then sends the results back to the MCP client.
MCP Orchestration
The MCP client, or orchestration layer, is a standalone component that runs its own server and can send requests to various MCP servers, whether they’re running locally or in an external environment. This is also the part the user directly interacts with. Classic examples include tools like Claude Desktop and Cursor.
If you've worked in the Generative AI or NLP space, you probably know that LLMs can't actually do things; they simply predict the next word based on the previous ones, a process called autoregressive modeling. So how does something like Cursor read a file, inspect its contents, or even edit it when you ask? That's where MCP orchestration comes in. We'll understand this better through the application we're building.
Calculator and converter MCP
To keep things simple and focus on the core concept, I chose to build basic MCP servers like a calculator and a converter. These servers are easier to implement, and since LLMs typically struggle with calculations, offloading such tasks to MCP servers makes sense.
Calculator MCP
Our calculator is a simple server that has functions/tools to do calculations like adding, subtracting, multiplying, etc.
Converter MCP
Our converter is simple too, and has tools to convert temperature.
MCP Client
Our client is a server that hosts an open-source model called Gorilla-openfunctions that has been finetuned on toolchain calling datasets. You can find the huggingface repo here.
What does the flow look like?.
The capabilities or functions offered by an MCP server are usually called tools.
MCP tools discovery
Each MCP server has something called a toolchain registry. When an MCP client starts up, it sends a toolchain discovery request to all the MCP servers it can access. In response, each MCP server returns metadata about its tools - such as the tool or function name, the parameters it accepts, the output it returns, and what it actually does. With this information, the MCP client knows exactly which tool to call when it needs to perform a specific task.
Now MCP Client/LLM will be aware of the tools available at its disposal.
MCP tool calling
As discussed above, an LLM can only generate text - it can't directly call any tools. So if a user asks the MCP client to add two numbers, the orchestration layer prepares a prompt that includes the user's query along with a list of available tools. It also instructs the model to use a specific syntax, like <call_function>function_name</call_function> with the required parameters, whenever a tool needs to be invoked. Since the LLM is trained on tool-calling examples, it knows when to trigger a tool call and when not to. The orchestration layer then watches the model's output; if it detects a tool-call pattern, it executes the tool and feeds the tool's result back into the LLM on the next step.
The idea is pretty simple, but what happens when there are lots of tools or servers available? After all, an LLM has a limited context window.
I had the same question when I was building the MCP client. There are several ways to handle this, such as:
(i) Storing all tool metadata in a vector database and retrieving the most relevant ones using similarity search.
(ii) Getting summaries of tools from each MCP server and letting the MCP orchestration handle routing with the help of a smaller router model.
(iii) Storing keywords or intents for each tool and selecting the right one based on those.
These are just a few options, but I decided to go with the first approach.
Enough with theory, let's implement it in code.
Building the MCP servers
We'll be implementing our MCP servers using a Python library called ModelContextProtocol (mcp).
Now, I know what you might be thinking: "Didn't you say you were going to build everything from scratch?"
You're right - and I actually did try building a fully custom version using FastAPI. It worked, but it was very basic. The overall flow was similar to what we discussed earlier, but I eventually decided to switch back to the MCP library for the server implementation for a few reasons:
(i) There are already plenty of MCP server implementations available online, and even the ones that run inside Docker or Claude Desktop give you a solid reference point.
(ii) There are far fewer MCP client or orchestration implementations - especially simple, well-explained ones - and that's where I want to focus.
(iii) I'm currently exploring some interesting ideas, and I want my setup to easily integrate with existing Docker-based MCP servers so I can build something useful for myself and the community.
If you still want the FastAPI version, I can share the code or even write a separate blog post about it. Just let me know in the comments.
With that out of the way, let's jump into the implementation.
Calculator MCP
# Calculator MCP server.
import os
import logging
from typing import Any
from mcp.server.fastmcp import FastMCP
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-server")
# Read host/port from environment (with defaults)
HOST = os.getenv("HOST", "0.0.0.0")
PORT = int(os.getenv("PORT", "8000"))
mcp = FastMCP(host=HOST, port=PORT)
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
@mcp.tool()
def subtract(a: int, b: int) -> int:
"""Subtract two numbers."""
return a - b
@mcp.tool()
def multiply(a: int, b: int) -> int:
"""Multiply two numbers."""
return a * b
@mcp.tool()
def divide(a: float, b: float) -> float:
"""Divide two numbers. Returns a float. Raises error if dividing by zero."""
if b == 0:
raise ValueError("Division by zero is not allowed.")
return a / b
if __name__ == "__main__":
logger.info("Starting MCP server on %s:%d", HOST, PORT)
# Running with SSE as transport because we will use same in client to interact.
mcp.run("sse") In the above, we are using the FastMCP class from the mcp library. The @mcp.tool decorators are used to define tools in the MCP library. The MCP library uses the comments and parameter names and their types to generate the metadata, which is sent to the MCP Client when it requests tools discovery. So it's important to document the functions you are defining in the tool.
Converter MCP
# Converter MCP server.py
import os
import logging
from typing import Any
from mcp.server.fastmcp import FastMCP
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-server")
# Read host/port from environment (with defaults)
HOST = os.getenv("HOST", "0.0.0.0")
PORT = int(os.getenv("PORT", "8000"))
mcp = FastMCP(host=HOST, port=PORT)
@mcp.tool()
def celsius_to_fahrenheit(celsius: float):
"""Convert temperature from Celsius to Fahrenheit."""
return (celsius * 9 / 5) + 32
@mcp.tool()
def fahrenheit_to_celsius(fahrenheit: float):
"""Convert temperature from Fahrenheit to Celsius."""
return (fahrenheit - 32) * 5 / 9
if __name__ == "__main__":
logger.info("Starting MCP server on %s:%d", HOST, PORT)
# FastMCP's .run() will start the server. transport "sse" is common for remote clients.
mcp.run("sse")
Similarly, we've defined our converter MCP server as well. Just like before, we're using SSE as the transport mechanism. The available options are stdout, HTTP, and SSE, but I'm choosing SSE because it's efficient and can communicate with remote MCP servers easily. This flexibility will likely be useful for me for the project I am planning next. The ports are the same in both the calculator and the converter this I will explain later why I chose to do.
Building MCP Client Orchestration
In my opinion, this is the most important part of the entire blog - the implementation of the MCP client.
Our MCP client will be built using FastAPI because it delivers great performance, is easy to work with, and makes deployment straightforward if we ever need to scale or expose the client as a service. Plus, FastAPI is quite popular right now, so chances are most readers will already be familiar with it.
Some utility functions
import asyncio
import json
import re
import ast
from typing import Dict, List, Optional, Union, Any,Tuple
import aiohttp
from urllib.parse import urljoin
from utils import *
import numpy as np
import copy
def build_tool_index(server_map: Dict[str, List[Dict]]) -> Dict[str, List[str]]:
"""
Build an index mapping tool name -> list of server URLs that provide that tool.
"""
index: Dict[str, List[str]] = {}
for server_url, tools in server_map.items():
for tool in tools:
name = tool.get("name")
if not name:
continue
index.setdefault(name, []).append(server_url)
return index
def try_parse_function_call(text: str) -> Optional[Tuple[str, dict]]:
"""
Robust function call extractor.
Handles noisy model outputs like:
<<<CALL_FUNCTION>>>subtract(a=2, b=3)
<<function>> add(x=1,y=2)
function_call: multiply(a=2)
"""
if not text:
return None
txt = text.strip()
# Flexible patterns
patterns = [
r"<+ *CALL_FUNCTION *>+\s*([A-Za-z0-9_]+)\s*\((.*)\)",
r"<+ *function *>+\s*([A-Za-z0-9_]+)\s*\((.*)\)",
r"CALL_FUNCTION\s*:?\s*([A-Za-z0-9_]+)\s*\((.*)\)",
r"([A-Za-z0-9_]+)\s*\((.*)\)"
]
for p in patterns:
m = re.search(p, txt, flags=re.IGNORECASE | re.DOTALL)
if not m:
continue
func_name = m.group(1)
args_str = m.group(2).strip()
args = {}
# No args
if args_str == "":
return func_name, args
# Split arguments by commas NOT inside quotes
parts = [s.strip() for s in re.split(r',(?=(?:[^"]*"[^"]*")*[^"]*$)', args_str) if s.strip()]
for part in parts:
if "=" in part:
k, v = part.split("=", 1)
k = k.strip()
v = v.strip()
try:
parsed_v = ast.literal_eval(v)
except Exception:
parsed_v = v.strip('"').strip("'")
args[k] = parsed_v
else:
# positional args
args.setdefault("_args", []).append(part)
return func_name, args
return None
def get_server_for_tool(
tool_name: str,
server_map: Dict[str, List[Dict]],
*,
first_only: bool = True,
case_sensitive: bool = False
) -> Optional[Union[str, List[str]]]:
"""
Return the server URL(s) that provide the tool `tool_name`.
Args:
tool_name: name of the tool to look up (e.g. "add").
server_map: dict mapping server_url -> list of tool dicts (like your server_map_dict).
first_only: if True (default) return the first matching URL (string).
if False return a list of all matching URLs (could be empty).
case_sensitive: if False (default) match tool name case-insensitively.
Returns:
If first_only=True -> a string (server URL) or None if not found.
If first_only=False -> list of matching URLs (empty list if none found).
"""
# Build index (fast for repeated lookups)
index = build_tool_index(server_map)
if not case_sensitive:
# build a case-insensitive mapping
ci_index = {}
for name, urls in index.items():
ci_index[name.lower()] = ci_index.get(name.lower(), []) + urls
matches = ci_index.get(tool_name.lower(), [])
else:
matches = index.get(tool_name, [])
if first_only:
return matches[0] if matches else None
return matches
def sanitize_for_json(obj: Any) -> Any:
"""
Recursively convert NumPy arrays/scalars and other common non-JSON types
to JSON-native Python types.
- numpy.ndarray -> list
- numpy scalar (np.generic) -> native python via .item()
- bytes/bytearray -> utf-8 str (or list of ints if decode fails)
"""
# numpy array -> list
if isinstance(obj, np.ndarray):
return obj.tolist()
# numpy scalar -> native Python (works across numpy versions)
if isinstance(obj, np.generic):
try:
return obj.item()
except Exception:
# fallback: convert to Python scalar string
return str(obj)
# bytes -> try decode
if isinstance(obj, (bytes, bytearray)):
try:
return obj.decode("utf-8")
except Exception:
return list(obj)
# dict/list/tuple -> recurse
if isinstance(obj, dict):
return {sanitize_for_json(k): sanitize_for_json(v) for k, v in obj.items()}
if isinstance(obj, list):
return [sanitize_for_json(v) for v in obj]
if isinstance(obj, tuple):
return tuple(sanitize_for_json(v) for v in obj)
# fallback: assume JSON-serializable already
return obj
def get_prompt(user_query: str, functions: list = []) -> str:
system = (
"You are a LLM you can answer questions. If you are not sure about it and you believe you can use a tool then you can call it.\n"
"If you need to call a tool, output EXACTLY the marker >>>CALL_FUNCTION<<< followed\n"
"by a single JSON object with keys 'name' and 'arguments' on the next line.\n"
"Example:\n"
">>>CALL_FUNCTION<<<\n"
"{\"name\": \"get_current_weather\", \"arguments\": {\"location\": \"Boston, MA\", \"unit\": \"celsius\"}}\n"
"If you do not need to call a function, just answer normally.\n",
"Also don't give example how to call the function but give the exact marker defined above."
)
functions_string = json.dumps(functions, indent=2)
return f"{system}\nAvailable Functions:\n{functions_string}\nUser: {user_query}\nAssistant:"
def _find_json_block(s: str):
"""
Locate the first JSON object in the string by finding the first '{' and
scanning forward until the braces are balanced. Handles quoted strings
and escaped quotes inside those strings.
"""
start = s.find('{')
if start == -1:
return None
depth = 0
in_string = None
escape = False
for i in range(start, len(s)):
ch = s[i]
if in_string is not None:
# We're inside a quoted string
if escape:
# previous char was backslash, so this char is escaped
escape = False
elif ch == "\\":
# start an escape for the next char
escape = True
elif ch == in_string:
# closing quote
in_string = None
# otherwise remain inside the string
else:
# not inside a string
if ch == '"' or ch == "'":
in_string = ch
elif ch == '{':
depth += 1
elif ch == '}':
depth -= 1
if depth == 0:
return s[start:i+1]
return None
def sanitize_for_json(obj: Any) -> Any:
"""
Recursively convert numpy arrays/scalars and basic non-JSON types to JSON-native types.
"""
# numpy arrays -> lists
if isinstance(obj, np.ndarray):
return obj.tolist()
# numpy scalars -> native Python types
if isinstance(obj, (np.integer, np.int_)):
return int(obj)
if isinstance(obj, (np.floating, np.float64)):
return float(obj)
if isinstance(obj, (np.bool_)):
return bool(obj)
# bytes -> decode if possible
if isinstance(obj, (bytes, bytearray)):
try:
return obj.decode("utf-8")
except Exception:
return list(obj)
# dict/list/tuple -> recurse
if isinstance(obj, dict):
return {sanitize_for_json(k): sanitize_for_json(v) for k, v in obj.items()}
if isinstance(obj, list):
return [sanitize_for_json(v) for v in obj]
if isinstance(obj, tuple):
return tuple(sanitize_for_json(v) for v in obj)
# fallback: leave as-is
return obj
def _split_args_top_level(args_str: str):
# split on commas that are not inside quotes or parentheses
parts = re.split(r',(?=(?:[^\"\']*[\"\'][^\"\']*[\"\'])*[^\"\']*$)', args_str)
# the regex above is a heuristic - trim parts
return [p.strip() for p in parts if p.strip()]
async def call_mcp_sse(name: str, arguments: dict, url: str = "http://localhost:8000/sse",
init_timeout: int = 10, call_timeout: int = 60):
"""
MCP SSE flow with a properly populated `initialize` payload.
Steps:
1) GET /sse and wait for event: endpoint -> data: /messages/?session_id=...
2) POST initialize (id=0) with required params {protocolVersion, capabilities, clientInfo}
and wait for its JSON-RPC response on the SSE stream.
3) POST tools/call (id=1)
4) Read SSE until JSON-RPC response for id=1 which contains the tool result
"""
rpc_initialize_id = 0
rpc_call_id = 1
# <-- IMPORTANT: supply required fields under params -->
init_params = {
"protocolVersion": "1.0",
# Capabilities: adjust as needed for your client. Keep keys conservative.
"capabilities": {
"supportsToolCalls": True,
"supportsStreaming": True,
# add other flags your client supports if desired
},
"clientInfo": {
"name": "simple-python-client",
"version": "0.1",
"description": "Debug client for MCP SSE calls"
}
}
init_payload = {"jsonrpc": "2.0", "id": rpc_initialize_id, "method": "initialize", "params": init_params}
call_payload = {"jsonrpc": "2.0", "id": rpc_call_id, "method": "tools/call", "params": {"name": name, "arguments": arguments}}
async def _sse_lines(resp):
while True:
line = await resp.content.readline()
if not line:
break
yield line.decode("utf-8").rstrip("\r\n")
headers = {"Accept": "text/event-stream"}
async with aiohttp.ClientSession() as session:
# 1) Open SSE GET
async with session.get(url, headers=headers, timeout=init_timeout) as resp:
if resp.status != 200:
body = await resp.text()
raise RuntimeError(f"SSE GET failed: status={resp.status}, body={body}")
# read until endpoint event
buf_event = None
buf_data_lines = []
messages_path = None
async for raw in _sse_lines(resp):
s = raw
if s == "":
if buf_event is not None:
data_str = "\n".join(buf_data_lines)
if buf_event == "endpoint":
messages_path = data_str.strip()
break
buf_event = None
buf_data_lines = []
continue
if s.startswith("event:"):
buf_event = s[len("event:"):].strip()
elif s.startswith("data:"):
buf_data_lines.append(s[len("data:"):].lstrip())
else:
# ignore other lines
pass
if not messages_path:
raise RuntimeError("Did not receive endpoint event from SSE GET (no /messages path).")
full_messages_url = urljoin(url, messages_path)
# 2) POST initialize
async with session.post(full_messages_url, json=init_payload, headers={"Content-Type": "application/json"}, timeout=10) as post_init:
if post_init.status not in (200, 202):
txt = await post_init.text()
raise RuntimeError(f"POST initialize failed: status={post_init.status}, body={txt}")
# server accepted initialize; wait for JSON-RPC response via SSE
# wait for initialize response (id == rpc_initialize_id)
init_deadline = asyncio.get_event_loop().time() + init_timeout
buf_event = None
buf_data_lines = []
init_ok = False
init_response = None
async for raw in _sse_lines(resp):
now = asyncio.get_event_loop().time()
if now > init_deadline:
break
s = raw
if s == "":
if buf_event is not None:
data_str = "\n".join(buf_data_lines)
try:
parsed = json.loads(data_str)
except Exception:
parsed = None
if isinstance(parsed, dict) and parsed.get("id") == rpc_initialize_id:
init_response = parsed
init_ok = True
break
buf_event = None
buf_data_lines = []
continue
if s.startswith("event:"):
buf_event = s[len("event:"):].strip()
elif s.startswith("data:"):
buf_data_lines.append(s[len("data:"):].lstrip())
if not init_ok:
# dump a helpful error
raise RuntimeError(f"Did not receive initialize response before timeout. Last init_response: {init_response}")
# Optional: inspect the initialize result for diagnostics
# print("Initialize response:", json.dumps(init_response, indent=2))
# 3) POST tools/call
async with session.post(full_messages_url, json=call_payload, headers={"Content-Type": "application/json"}, timeout=10) as call_resp:
if call_resp.status not in (200, 202):
txt = await call_resp.text()
raise RuntimeError(f"POST tools/call failed: status={call_resp.status}, body={txt}")
# 4) wait for tools/call response (id == rpc_call_id)
call_deadline = asyncio.get_event_loop().time() + call_timeout
buf_event = None
buf_data_lines = []
async for raw in _sse_lines(resp):
now = asyncio.get_event_loop().time()
if now > call_deadline:
break
s = raw
if s == "":
if buf_event is not None:
data_str = "\n".join(buf_data_lines)
try:
parsed = json.loads(data_str)
except Exception:
parsed = None
if isinstance(parsed, dict) and parsed.get("id") == rpc_call_id and "result" in parsed:
res = parsed["result"]
# try to extract text content
if isinstance(res, dict) and "content" in res:
texts = []
for item in res["content"]:
if isinstance(item, dict) and item.get("type") == "text" and "text" in item:
texts.append(item["text"])
return "\n".join(texts) if texts else json.dumps(res)
if isinstance(res, str):
return res
return json.dumps(res)
buf_event = None
buf_data_lines = []
continue
if s.startswith("event:"):
buf_event = s[len("event:"):].strip()
elif s.startswith("data:"):
buf_data_lines.append(s[len("data:"):].lstrip())
raise RuntimeError("Timed out waiting for tools/call JSON-RPC response over SSE.")
These are some of the key utility functions used in our MCP client. Although most of them are well documented, I'll highlight the ones that matter the most.
The call_mcp_sse function sends requests to the MCP server using the SSE transport, triggers the appropriate tools, and returns their results. The get_prompt function prepares the prompt for our LLM, instructing it to generate a specific function-call signature when it decides a tool should be invoked. This prompt is something I arrived at through experimentation-it works well for now, but can be refined further in the future, possibly through more experimentation or by using frameworks like DsPy.
Finally, the get_server_for_tool function maps a tool name to the correct MCP server URL responsible for hosting it. This file is referred to as utils.py in the code.
RAG Tool retrieval module
As discussed in the Core Concepts section, LLMs operate within a fixed context window. To overcome this limitation, we use a RAG-based tool-metadata retrieval system that helps identify the tools most relevant to the current request.
# mcp_tools_ret_utils.py (replacement functions)
import json
import os
from typing import Dict, List, Any
import numpy as np
from sentence_transformers import SentenceTransformer
import lancedb
from lancedb import connect
# Optional imports for typed vector schema fallback
try:
from lancedb.pydantic import Vector, LanceModel # newer versions
HAVE_PYDANTIC = True
except Exception:
HAVE_PYDANTIC = False
EMBED_MODEL_NAME = "all-MiniLM-L6-v2" # change if you use another embedder
def _make_tool_text(tool: Dict[str, Any]) -> str:
name = tool.get("name", "")
desc = tool.get("description", "") or ""
params = tool.get("parameters", None)
try:
params_str = json.dumps(params, ensure_ascii=False)
except Exception:
params_str = str(params)
return f"{name}. {desc}. parameters: {params_str}"
def _ensure_json_schema(params: Any) -> Dict[str, Any]:
if isinstance(params, dict):
return params
if isinstance(params, str):
try:
parsed = json.loads(params)
if isinstance(parsed, dict):
return parsed
except Exception:
pass
return {"type": "object", "properties": {}}
def print_lancedb_debug(db):
print("lancedb version:", getattr(lancedb, "__version__", "unknown"))
print("db object methods:", [m for m in dir(db) if not m.startswith("_")])
def index_tools_to_lancedb(
server_map_dict: Dict[str, List[Dict[str, Any]]],
db_path: str = "./mcp_tools_lancedb",
table_name: str = "mcp_tools",
embedding_model_name: str = EMBED_MODEL_NAME,
overwrite: bool = False,
) -> None:
"""
Index discovered tools into LanceDB in a portable way that works across versions.
The vector column will be stored under the key "embedding" (list[float]).
"""
model = SentenceTransformer(embedding_model_name)
rows = []
for server_url, tools in server_map_dict.items():
for i, tool in enumerate(tools):
name = tool.get("name") or f"tool_{i}"
doc_id = f"{server_url}::{name}::{i}"
params = _ensure_json_schema(tool.get("parameters"))
text = _make_tool_text({"name": name, "description": tool.get("description", ""), "parameters": params})
rows.append(
{
"id": doc_id,
"server_url": server_url,
"tool_name": name,
"description": tool.get("description", "") or "",
"parameters": params,
"text": text,
# embedding is added below
}
)
if not rows:
print("[index_tools_to_lancedb] no tools to index.")
return
texts = [r["text"] for r in rows]
embeddings = model.encode(texts, show_progress_bar=False, convert_to_numpy=True).astype(np.float32)
for r, emb in zip(rows, embeddings):
# Convert to Python list so older lancedb versions accept it
r["embedding"] = emb.tolist()
os.makedirs(db_path, exist_ok=True)
db = connect(db_path)
# debug info (uncomment if you want to inspect)
# print_lancedb_debug(db)
# Try to remove existing table in a portable way
if overwrite:
# Some versions expose drop_table or delete_table
if hasattr(db, "drop_table"):
try:
db.drop_table(table_name)
print(f"[index_tools_to_lancedb] dropped existing table '{table_name}' (overwrite=True).")
except Exception:
pass
elif hasattr(db, "delete_table"):
try:
db.delete_table(table_name)
print(f"[index_tools_to_lancedb] deleted existing table '{table_name}' (overwrite=True).")
except Exception:
pass
# Try several create_table signatures used across releases
created = False
create_exceptions = []
try:
# common simple form: create_table(name, data)
db.create_table(table_name, rows)
created = True
except Exception as e1:
create_exceptions.append(e1)
try:
# alternate: create_table(name=..., data=...)
db.create_table(name=table_name, data=rows)
created = True
except Exception as e2:
create_exceptions.append(e2)
# fallback to explicit schema using pydantic Vector (if available)
if HAVE_PYDANTIC:
try:
# determine embedding dim from first row
dim = len(embeddings[0])
class ToolModel(LanceModel):
id: str
server_url: str
tool_name: str
description: str
parameters: dict
text: str
embedding: Vector(dim)
# create table with schema and mode overwrite to replace if exists
db.create_table(table_name, schema=ToolModel, mode="overwrite")
tbl = db.open_table(table_name)
# many versions expose add/insert or append, try common names
if hasattr(tbl, "add"):
tbl.add(rows)
elif hasattr(tbl, "insert"):
tbl.insert(rows)
elif hasattr(tbl, "append"):
tbl.append(rows)
else:
# as last resort, try db.create_table with data after schema create
try:
db.create_table(table_name, rows)
except Exception:
raise RuntimeError("Could not add rows to newly created typed table.")
created = True
except Exception as e3:
create_exceptions.append(e3)
if not created:
# If we still didn't create, try opening existing table and inserting rows
try:
tbl = db.open_table(table_name)
# Try common insertion methods
if hasattr(tbl, "add"):
tbl.add(rows)
elif hasattr(tbl, "insert"):
tbl.insert(rows)
elif hasattr(tbl, "append"):
tbl.append(rows)
else:
raise RuntimeError("Opened table exists but doesn't support add/insert/append.")
created = True
print(f"[index_tools_to_lancedb] appended {len(rows)} rows to existing table '{table_name}'.")
except Exception as e_open:
create_exceptions.append(e_open)
# Give a helpful error message
msg = "Failed to create or open LanceDB table. Errors:\n" + "\n".join(
f"- {type(x).__name__}: {x}" for x in create_exceptions
)
raise RuntimeError(msg)
print(f"[index_tools_to_lancedb] table '{table_name}' ready with {len(rows)} rows.")
def fetch_top_k_tools_formatted(
query: str,
db_path: str = "./mcp_tools_lancedb",
table_name: str = "mcp_tools",
k: int = 5,
embedding_model_name: str = EMBED_MODEL_NAME
) -> List[Dict[str, Any]]:
"""
Query LanceDB and return top-k tools in the requested schema.
Uses the portable search pattern: table.search(vector).limit(k).to_pandas()
"""
model = SentenceTransformer(embedding_model_name)
q_emb = model.encode([query], convert_to_numpy=True)[0].astype(np.float32).tolist()
db = connect(db_path)
try:
table = db.open_table(table_name)
except Exception as e:
raise RuntimeError(f"Could not open LanceDB table '{table_name}': {e}") from e
# Portable search: call search(q_emb) then chain limit() and to_pandas() if available
try:
search_obj = table.search(q_emb)
# chain limit if method exists
if hasattr(search_obj, "limit"):
search_obj = search_obj.limit(k)
# prefer to_pandas when available
if hasattr(search_obj, "to_pandas"):
df = search_obj.to_pandas()
hits = []
for _, row in df.iterrows():
params = _ensure_json_schema(row.get("parameters", None))
entry = {"name": row.get("tool_name"), "description": row.get("description") or "", "parameters": params}
entry["server_url"] = row.get("server_url")
hits.append(entry)
return hits
else:
# try execute() or to_list
try:
executed = search_obj.execute()
except Exception:
executed = list(search_obj)
hits = []
for hit in executed[:k]:
params = _ensure_json_schema(hit.get("parameters", None))
entry = {"name": hit.get("tool_name"), "description": hit.get("description") or "", "parameters": params}
entry["server_url"] = hit.get("server_url")
hits.append(entry)
return hits
except Exception as e:
raise RuntimeError(f"Search failed on table '{table_name}': {e}") from e
In the above code, we have implemented a simple module using MiniLM as the embedding model. This module has functions to store our tools metadata in a vector database called lancedb. And retrieve the top K that are relevant to our query. This file is referred to as mcp_tools_ret_utils.py in the code.
Main MCP client code
Importing libraries
import asyncio
import json
from typing import List, Dict, Any, Optional,Tuple
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from llama_cpp import Llama
from llama_index.tools.mcp import BasicMCPClient, McpToolSpec
from utils import sanitize_for_json,get_prompt, call_mcp_sse,get_server_for_tool,try_parse_function_call
from mcp_tools_ret_utils import index_tools_to_lancedb, fetch_top_k_tools_formatted
import re
import os
Some variable declarations
MCP_SERVER_URLS = os.environ.get('MCP_SERVER_URLS', '')
if MCP_SERVER_URLS:
MCP_SERVER_URLS = MCP_SERVER_URLS.split(',')
MODEL_PATH = "gorilla-openfunctions-v2-GGUF/gorilla-openfunctions-v2-q4_K_M.gguf"
app = FastAPI(title="MCP")
# Globals populated at startup
server_map_dict: Dict[str, List[Dict[str, Any]]] = {}
llm: Optional[Llama] = None
class QueryRequest(BaseModel):
query: str
k: int = 4
Main code block
async def discover_tools(server_urls: List[str]) -> Dict[str, List[Dict[str, Any]]]:
"""
Discover tools on each MCP server URL and return a mapping:
{ server_url: [ {name, description, parameters}, ... ], ... }
This is adapted from your provided snippet; it is defensive about missing metadata.
"""
server_map_dict_local: Dict[str, List[Dict[str, Any]]] = {}
for server_url in server_urls:
functions: List[Dict[str, Any]] = []
try:
client = BasicMCPClient(server_url)
tool_spec = McpToolSpec(client=client)
# Wait for tool list asynchronously with a timeout
tools = await asyncio.wait_for(tool_spec.to_tool_list_async(), timeout=10)
for tool in tools:
meta = getattr(tool, "_metadata", None)
if meta is None:
# fallback: try to access tool metadata properties defensively
name = getattr(tool, "name", None) or getattr(tool, "fn", None) or "unknown"
description = getattr(tool, "description", "") or ""
params_raw = getattr(tool, "fn_schema_str", None)
else:
name = getattr(meta, "name", None) or "unknown"
description = getattr(meta, "description", "") or ""
params_raw = getattr(meta, "fn_schema_str", None)
# Try to parse parameters if it's a JSON string; otherwise keep as-is
params = None
try:
if isinstance(params_raw, str):
params = json.loads(params_raw)
else:
params = params_raw
except Exception as e:
print(f"[discover_tools] Warning: failed to parse fn_schema_str for tool '{name}': {e}")
params = params_raw
func_dict = {
"name": name,
"description": description,
"parameters": params,
}
functions.append(func_dict)
except asyncio.TimeoutError:
print(f"[discover_tools] Timeout while fetching tools from {server_url}")
except Exception as e:
print(f"[discover_tools] Error while contacting {server_url}: {e}")
# Always set an entry (possibly empty) for this server_url
server_map_dict_local[server_url] = functions
return server_map_dict_local
@app.on_event("startup")
async def startup_event():
"""Discover tools and index them at startup. Also create the Llama instance once."""
global server_map_dict, llm
# Discover MCP tools (async)
try:
server_map_dict = await discover_tools(MCP_SERVER_URLS)
print("Discovered tools:", {k: len(v) for k, v in server_map_dict.items()})
except Exception as e:
print("Warning: discover_tools failed at startup:", e)
# Index tools into lancedb (safe to call even if server_map_dict is empty)
try:
index_tools_to_lancedb(server_map_dict, db_path="./mcp_tools_lancedb", overwrite=True)
print("Indexed tools to lancedb")
except Exception as e:
print("Warning: indexing to lancedb failed:", e)
# Instantiate Llama in a threadpool to avoid blocking the event loop
loop = asyncio.get_event_loop()
def _create_llm():
print("Creating Llama model (this may take a while)...")
return Llama(model_path=MODEL_PATH, n_threads=8, n_ctx=2048, n_gpu_layers=35)
try:
llm = await loop.run_in_executor(None, _create_llm)
print("Llama model created")
except Exception as e:
print("Error creating Llama model:", e)
llm = None
print(f"server_map_dict={server_map_dict}")
async def _call_llm(prompt: str, max_tokens: int = 2048) -> str:
"""Run the synchronous llm(prompt) in a threadpool and return the text string."""
global llm
if llm is None:
raise RuntimeError("LLM not initialized")
loop = asyncio.get_event_loop()
def _sync_call():
out = llm(prompt, max_tokens=max_tokens, echo=False)
if isinstance(out, dict):
return out.get("choices", [{}])[0].get("text") or out.get("text") or str(out)
return str(out)
return await loop.run_in_executor(None, _sync_call)
@app.post("/query")
async def run_query(req: QueryRequest):
"""Main endpoint. Accepts JSON: {"query":"Add two numbers 3 and 4.", "k":2}
Flow:
- fetch top-k tools from lancedb using fetch_top_k_tools_formatted
- build prompt, call Llama
- detect function call; if found, call the MCP tool via SSE and optionally follow up
- return JSON with result and any tool output
"""
try:
# Step 1: retrieve candidate tools from lancedb
functions = fetch_top_k_tools_formatted(req.query, k=req.k)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error fetching tools from index: {e}")
print(f"functions returned by discovery={functions}")
# Step 2: prepare prompt
#functions_no_internal = strip_internal_fields(functions, fields=("embedding",))
safe_functions = sanitize_for_json(functions)
prompt = get_prompt(req.query, safe_functions)
# Step 3: call LLM
try:
raw_text = await _call_llm(prompt, max_tokens=2048)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error during LLM call: {e}")
# Step 4: detect function call
call = try_parse_function_call(raw_text)
if call is None:
# No function call detected - return raw LLM output
return {
"final_text": raw_text,
"tool_called": None,
"tool_result": None,
"raw_model_output": raw_text,
"error": "No valid function call detected in model output"
}
func_name, func_args = call
# Step 5: execute MCP tool via SSE
url_to_call=get_server_for_tool(func_name, server_map_dict)
try:
tool_result = await call_mcp_sse(func_name, func_args,url=url_to_call)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error calling MCP tool: {e}")
# Step 6: Optional follow-up LLM call to compose final answer
followup_prompt = (
prompt
+ "\n\n(The tool call has been executed.)\n"
+ f"Tool name: {func_name}\nTool output: {tool_result}\n" +
"Now the result need to be formatted in a human readable tone."
+ "Assistant:"
)
try:
final_text = await _call_llm(followup_prompt, max_tokens=2048)
except Exception as e:
# If follow-up LLM fails, still return tool_result and raw_text
print(f"Exception={str(e)}")
return {
"final_text": None,
"tool_called": func_name,
"tool_result": sanitize_for_json(tool_result),
"raw_model_output": raw_text,
"followup_error": str(e),
}
return {
"final_text": final_text,
"tool_called": func_name,
"tool_result": sanitize_for_json(tool_result),
"raw_model_output": raw_text,
}
if __name__ == '__main__':
import uvicorn
# Run with: python fastapi_mcp_endpoint.py OR: uvicorn fastapi_mcp_endpoint:app --reload
uvicorn.run(app, host='0.0.0.0', port=3000, log_level='info')
There is a startup event declared called startup_event, which is called upon when the application is launched. This event calls the function called discover_tools, which would iteratively discover tools for the server in the server_urls URL list. The retrieved tools are stored in the dictionary server_map_dict. This is stored/indexed in our vector DB using the function index_tools_to_lancedb. The LLM is loaded next in a non-blocking manner using the asyncio threadpool. It might be useful in real-world applications where you don't want the loading to interfere with the server loading, as model loading is CPU-heavy.
The run query function in the query endpoint is the most important in the entire blog. Again its well documented with steps to follow. At the crux, it does the following operations:
(i) Get the top tools suitable for the query using RAG from the vector DB.
(ii) Embed it in a prompt using get_prompt and pass to LLM.
(iii) Extract the function's signature if tool calling is needed.
(iv) Call the appropriate tool and get results.
(v) Reformat the result and present to the user, recall another tool if needed.
Packaging the MCP client and server as Docker
Nothing is stopping us from manually setting up the dependencies for the MCP server and MCP clients and running all three as separate scripts. That alone would make the MCP client–server interface work. However, containerizing the setup with Docker makes dependency management much easier, and it also simplifies migrating the entire system to another machine.
For example, I develop on an Ubuntu VM because my main system runs Windows 11. With Docker, I can take the same setup and run it with GPU support directly on Windows without any additional configuration.
So I wrote a helper script that takes the folder structure in my project and iterates over the MCP server folder, and generates Docker-Compose YAML for running the container in the required order. Please refer to my git repo in the references section for finding and running that script.
MCP in action
Further Improvements
1. The retrieval logic for tools metadata is very basic and can be further refined.
2. Prompts can be improved as discussed above.
3. Better inference pipelines like vLLM could be used for inference.
4. The model can further be finetuned to improve the result, although prompt optimization should be prioritized here.
References and Further Readings
If you liked this blog, please consider sharing it with your network, which would motivate me to write more blogs like this. If you have any questions, please feel free to drop them in the comments section.
Thanks for taking the time to read the complete blog.
Comments
No comments yet. Be the first to comment!