MCP Nav Project – Starter Code (Spring Boot +
Python)
Copy these files into a fresh workspace. The Python service does routing + traffic simulation
(and exposes minimal MCP tools). Spring Boot calls the Python HTTP API for your demo.
0) Project Layout
mcx-nav/
├─ python-service/
│ ├─ [Link]
│ ├─ graph_data.py
│ ├─ [Link]
│ ├─ [Link]
│ ├─ [Link] # FastAPI HTTP server (Java calls this)
│ └─ mcp_server.py # Minimal MCP server exposing tools
└─ springboot-api/
├─ [Link]
└─ src/main/java/com/example/nav/
├─ [Link]
├─ config/[Link]
├─ dto/[Link]
├─ dto/[Link]
├─ dto/[Link]
└─ web/[Link]
1) Python Service
1.1 [Link]
fastapi==0.111.0
uvicorn==0.30.0
networkx==3.3
pydantic==2.7.1
mcp>=1.2.0
1
1.2 graph_data.py
# Builds a small demo graph (A..H). Distances are in km, speed in km/h.
# You can replace this with OSMnx later if you want real maps.
from __future__ import annotations
import networkx as nx
# Base graph with geometric distance (km)
def build_base_graph() -> [Link]:
G = [Link]()
nodes = {
"A": (0, 0), "B": (2, 0), "C": (4, 0), "D": (6, 0),
"E": (0, 2), "F": (2, 2), "G": (4, 2), "H": (6, 2),
}
for n, pos in [Link]():
G.add_node(n, pos=pos)
# Add edges with distance (km) and default speed (km/h)
def add(u, v, dist, speed=50.0):
G.add_edge(u, v, distance=dist, speed=speed)
# Horizontal
add("A","B",2); add("B","C",2); add("C","D",2)
add("E","F",2); add("F","G",2); add("G","H",2)
# Vertical
add("A","E",2); add("B","F",2); add("C","G",2); add("D","H",2)
# Diagonals / shortcuts
add("A","F",2.8); add("B","G",2.8); add("C","H",2.8)
return G
1.3 [Link]
from __future__ import annotations
import networkx as nx
from typing import Dict, Tuple
class TrafficModel:
"""Holds dynamic speeds per edge. Defaults to edge 'speed' attr if not
overridden."""
def __init__(self, G: [Link]):
self.G = G
[Link]: Dict[Tuple[str, str], float] = {}
2
def set_speed(self, u: str, v: str, speed_kmph: float):
key = tuple(sorted((u, v)))
[Link][key] = max(1.0, float(speed_kmph))
def get_speed(self, u: str, v: str) -> float:
key = tuple(sorted((u, v)))
if key in [Link]:
return [Link][key]
return float(self.G[u][v].get("speed", 50.0))
def edge_time_hours(self, u: str, v: str) -> float:
dist = float(self.G[u][v]["distance"]) # km
spd = self.get_speed(u, v) # km/h
return dist / spd
1.4 [Link]
from __future__ import annotations
import networkx as nx
from typing import Dict, List, Tuple
from traffic import TrafficModel
class Router:
def __init__(self, G: [Link], traffic: TrafficModel):
self.G = G
[Link] = traffic
def path_costs(self, path: List[str]) -> Tuple[float, float]:
"""Returns (distance_km, time_hours) for a path."""
dist = 0.0
time_h = 0.0
for u, v in zip(path[:-1], path[1:]):
dist += float(self.G[u][v]["distance"])
time_h += [Link].edge_time_hours(u, v)
return dist, time_h
def shortest_distance(self, s: str, t: str) -> Dict:
path = nx.shortest_path(self.G, s, t, weight=lambda u, v, d:
d["distance"]) # km
dist, time_h = self.path_costs(path)
return {"path": path, "distance_km": dist, "time_min": time_h * 60}
def shortest_time(self, s: str, t: str) -> Dict:
def time_weight(u, v, d):
dist = d["distance"]
3
spd = [Link].get_speed(u, v)
return dist / spd # hours
path = nx.shortest_path(self.G, s, t, weight=time_weight)
dist, time_h = self.path_costs(path)
return {"path": path, "distance_km": dist, "time_min": time_h * 60}
def tradeoff(self, s: str, t: str, alpha: float) -> Dict:
"""
alpha in [0,1]:
0.0 => minimize distance
1.0 => minimize time
cost = alpha*time_hours + (1-alpha)*distance_km (unitless mix)
"""
a = max(0.0, min(1.0, float(alpha)))
def mixed(u, v, d):
dist = float(d["distance"]) # km
time_h = [Link].edge_time_hours(u, v)
return a * time_h + (1 - a) * dist
path = nx.shortest_path(self.G, s, t, weight=mixed)
dist, time_h = self.path_costs(path)
return {
"path": path,
"distance_km": dist,
"time_min": time_h * 60,
"alpha": a
}
1.5 [Link] (FastAPI HTTP server)
from __future__ import annotations
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
from graph_data import build_base_graph
from traffic import TrafficModel
from router import Router
G = build_base_graph()
traffic = TrafficModel(G)
router = Router(G, traffic)
app = FastAPI(title="MCP Nav Python Service")
class RouteReq(BaseModel):
start: str
4
end: str
alpha: float = 0.5 # 0..1
class TrafficReq(BaseModel):
u: str
v: str
speed_kmph: float
@[Link]("/route")
async def compute_route(req: RouteReq):
s, t = [Link], [Link]
for n in (s, t):
if n not in [Link]:
raise HTTPException(400, f"Unknown node: {n}")
result = [Link](s, t, [Link])
return {"ok": True, "result": result}
@[Link]("/route/fast")
async def shortest_time(req: RouteReq):
s, t = [Link], [Link]
result = router.shortest_time(s, t)
return {"ok": True, "result": result}
@[Link]("/route/short")
async def shortest_dist(req: RouteReq):
s, t = [Link], [Link]
result = router.shortest_distance(s, t)
return {"ok": True, "result": result}
@[Link]("/traffic/update")
async def update_traffic(req: TrafficReq):
u, v, spd = req.u, req.v, req.speed_kmph
if not G.has_edge(u, v):
raise HTTPException(400, f"Unknown edge: {u}-{v}")
traffic.set_speed(u, v, spd)
return {"ok": True, "edge": [u, v], "speed_kmph": spd}
if __name__ == "__main__":
[Link](app, host="[Link]", port=8001)
1.6 mcp_server.py (Minimal MCP tools)
"""
Minimal MCP server exposing two tools so you can claim MCP support.
Run separately from [Link] when showcasing MCP integration with an MCP-enabled
client.
5
"""
from __future__ import annotations
import asyncio
from [Link] import FastMCP
from graph_data import build_base_graph
from traffic import TrafficModel
from router import Router
G = build_base_graph()
traffic = TrafficModel(G)
router = Router(G, traffic)
mcp = FastMCP("mcp-nav-server")
@[Link]()
async def get_route(start: str, end: str, alpha: float = 0.5) -> dict:
"""Return best route for start→end using trade-off alpha in [0,1]."""
return [Link](start, end, alpha)
@[Link]()
async def update_traffic(u: str, v: str, speed_kmph: float) -> dict:
"""Set dynamic speed (km/h) for edge u–v."""
if not G.has_edge(u, v):
return {"ok": False, "error": f"Unknown edge {u}-{v}"}
traffic.set_speed(u, v, speed_kmph)
return {"ok": True, "edge": [u, v], "speed_kmph": speed_kmph}
if __name__ == "__main__":
[Link]([Link]())
2) Spring Boot API (Java)
2.1 [Link]
<project xmlns="[Link] xmlns:xsi="[Link]
2001/XMLSchema-instance"
xsi:schemaLocation="[Link] http://
[Link]/xsd/[Link]">
<modelVersion>4.0.0</modelVersion>
<groupId>[Link]</groupId>
<artifactId>nav</artifactId>
<version>0.0.1-SNAPSHOT</version>
<properties>
<[Link]>17</[Link]>
6
<[Link]>3.3.1</[Link]>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>[Link]</groupId>
<artifactId>spring-boot-dependencies</artifactId>
<version>${[Link]}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>[Link]</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>[Link]</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<dependency>
<groupId>[Link]</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>[Link]</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
2.2 [Link]
package [Link];
import [Link];
import [Link];
@SpringBootApplication
public class NavApplication {
7
public static void main(String[] args) {
[Link]([Link], args);
}
}
2.3 config/[Link]
package [Link];
import [Link];
import [Link];
import [Link];
@Configuration
public class AppConfig {
@Bean
public RestClient restClient() {
return [Link]().baseUrl("[Link]
}
}
2.4 DTOs
dto/[Link]
package [Link];
import [Link].*;
public class RouteRequest {
@NotBlank
private String start;
@NotBlank
private String end;
@DecimalMin("0.0") @DecimalMax("1.0")
private Double alpha = 0.5; // 0..1
public String getStart() { return start; }
public void setStart(String start) { [Link] = start; }
public String getEnd() { return end; }
public void setEnd(String end) { [Link] = end; }
public Double getAlpha() { return alpha; }
public void setAlpha(Double alpha) { [Link] = alpha; }
}
8
dto/[Link]
package [Link];
import [Link];
public class RouteResponse {
private boolean ok;
private Result result;
public static class Result {
private List<String> path;
private double distance_km;
private double time_min;
private Double alpha;
public List<String> getPath() { return path; }
public void setPath(List<String> path) { [Link] = path; }
public double getDistance_km() { return distance_km; }
public void setDistance_km(double distance_km) { this.distance_km =
distance_km; }
public double getTime_min() { return time_min; }
public void setTime_min(double time_min) { this.time_min = time_min; }
public Double getAlpha() { return alpha; }
public void setAlpha(Double alpha) { [Link] = alpha; }
}
public boolean isOk() { return ok; }
public void setOk(boolean ok) { [Link] = ok; }
public Result getResult() { return result; }
public void setResult(Result result) { [Link] = result; }
}
dto/[Link]
package [Link];
import [Link].*;
public class TrafficUpdateRequest {
@NotBlank
private String u;
@NotBlank
private String v;
@Positive
private double speed_kmph;
9
public String getU() { return u; }
public void setU(String u) { this.u = u; }
public String getV() { return v; }
public void setV(String v) { this.v = v; }
public double getSpeed_kmph() { return speed_kmph; }
public void setSpeed_kmph(double speed_kmph) { this.speed_kmph =
speed_kmph; }
}
2.5 web/[Link]
package [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link].*;
import [Link];
@RestController
@RequestMapping("/api")
public class RouteController {
private final RestClient rest;
public RouteController(RestClient rest) {
[Link] = rest;
}
@PostMapping("/navigate")
public ResponseEntity<RouteResponse> navigate(@Validated @RequestBody
RouteRequest req) {
RouteResponse resp = [Link]()
.uri("/route")
.body(req)
.retrieve()
.body([Link]);
return [Link](resp);
}
@PostMapping("/traffic-update")
public ResponseEntity<String> traffic(@Validated @RequestBody
TrafficUpdateRequest req) {
String resp = [Link]()
10
.uri("/traffic/update")
.body(req)
.retrieve()
.body([Link]);
return [Link](resp);
}
}
3) How to Run (Local)
Terminal 1 – Python service
cd python-service
python -m venv .venv && source .venv/bin/activate #
Windows: .venv\Scripts\activate
pip install -r [Link]
python [Link] # FastAPI on [Link]
Terminal 2 – Spring Boot
cd springboot-api
mvn spring-boot:run # runs on [Link]
Test Calls
• Get route (alpha=0.6)
curl -X POST [Link] \
-H 'Content-Type: application/json' \
-d '{"start":"A","end":"H","alpha":0.6}'
• Slow down edge B–C to 10 km/h
curl -X POST [Link] \
-H 'Content-Type: application/json' \
-d '{"u":"B","v":"C","speed_kmph":10}'
• Re-run navigate and observe route change.
11
(Optional) Run MCP server (separate)
cd python-service
python mcp_server.py # starts MCP server over stdio for MCP-enabled clients
4) Notes for Your Report / Viva
• Trade-off formula: cost = α·time_hours + (1-α)·distance_km .
• Dynamic routing: Traffic updates change per-edge speeds → recompute shortest path by mixed
cost.
• Why Spring Boot + Python: Java layer exposes clean REST for enterprise flavor; Python handles
graph/optimization quickly.
• Upgrade path: Replace demo graph with OSMnx for a real city, or plug in A and contraction
hierarchies* for scale.
5) Quick Tasks You Can Add (If Time Permits)
• Add /api/route/short and /api/route/fast pass-through endpoints.
• Streamlit viz drawing the graph and highlighting the chosen path.
• Cache last N routes in Spring Boot and display metrics.
Good luck—ship it! 🚀
12