FastMCP Supply Chain Optimizer
A custom implementation of FastMCP (Model Context Protocol) for real-time supply chain optimization using Gemini AI. This project demonstrates low-latency, multi-tool orchestration inspired by Anthropic's internal FastMCP system.
🎯 What This Demonstrates
- Custom FastMCP Implementation: Multi-tool calling at every LLM processing step (not sequential)
- Real-time Event Processing: Stream of supply chain events with live AI responses
- Intelligent Recommendations: AI-powered inventory optimization with actionable insights
- Live Web Interface: Real-time monitoring and control with beautiful UI
- Modular Tool Architecture: Easy to extend and modify for different use cases
🔧 About FastMCP vs MCP
FastMCP is not open source - it's Anthropic's internal implementation. This project is a minimal simulation of FastMCP's key innovation:
Core Difference: Parallel Tool Calling
- Standard MCP: Sequential alternating between 1 LLM call → 1 tool call → 1 LLM call
- FastMCP: Multiple tools called at every step of LLM processing
- This Implementation: Simulates FastMCP's approach with multiple tool execution per event
FastMCP isn't open source, so I built a minimal simulation of a low-latency multi-tool orchestration stack inspired by it — showcasing how an LLM agent can respond to real-time supply chain updates with actionable suggestions via routed tools.
🚀 Quick Start
1. Install Dependencies
pip install -r requirements.txt
2. Run the Application
python3 flask_app.py
3. Open Browser
Navigate to http://localhost:5000
4. Alternative: Use Local LLM
For data privacy and internal tool usage, you can replace Gemini API with your own local LLM using local-llm-api:
# Clone and setup local LLM API
git clone https://github.com/ANSH-RIYAL/local-llm-api.git
cd local-llm-api
./run_server.sh
# Modify fastmcp_server.py to use local API instead of Gemini
# Replace GEMINI_API_KEY with CUSTOM_API_URL = "http://localhost:8050"
🎮 How to Use
- Start FastMCP Server: Click "Start FastMCP Server" to initialize the AI agent
- Start Event Stream: Click "Start Event Stream" to begin processing supply chain events
- Monitor Results: Watch the terminal output and action recommendations in real-time
- Stop When Done: Use the stop buttons to gracefully shut down
🛠️ Tools Implemented
Core Supply Chain Tools
1. get_inventory_status
- Purpose: Check current inventory levels across all warehouses
- Parameters:
product_id(optional) - Returns: Complete inventory data for product or all products
- Example:
{"product_id": "P001"}→ Returns warehouse A/B/C stock levels
2. update_inventory
- Purpose: Modify warehouse stock levels (add/subtract)
- Parameters:
product_id,warehouse,quantity - Returns: Success status and inventory change details
- Example:
{"product_id": "P001", "warehouse": "warehouse_A", "quantity": -10}
3. calculate_transfer
- Purpose: Move inventory between warehouses
- Parameters:
product_id,from_warehouse,to_warehouse,quantity - Returns: Transfer execution details and new inventory levels
- Example:
{"product_id": "P001", "from_warehouse": "warehouse_B", "to_warehouse": "warehouse_A", "quantity": 20}
4. predict_stockout
- Purpose: Forecast when products will run out of stock
- Parameters:
product_id,warehouse - Returns: Risk level and predicted stockout timeline
- Example:
{"product_id": "P001", "warehouse": "warehouse_A"}→ "HIGH risk, 1-2 days"
5. recommend_reorder
- Purpose: Suggest reorder quantities and suppliers
- Parameters:
product_id,quantity - Returns: Order details with cost calculations
- Example:
{"product_id": "P001", "quantity": 50}→ "ORDER: 50 units from Supplier X at $5.50/unit"
How to Modify Tools
Adding New Tools
- Add function to
supply_chain_tools.py:
def new_tool_function(self, param1: str, param2: int) -> Dict[str, Any]:
"""Description of what this tool does"""
# Implementation logic
return {"success": True, "result": "tool output"}
- Register tool in
fastmcp_server.py:
Tool(
name="new_tool_function",
description="Description of what this tool does",
inputSchema={
"type": "object",
"properties": {
"param1": {"type": "string", "description": "Parameter 1"},
"param2": {"type": "integer", "description": "Parameter 2"}
},
"required": ["param1", "param2"]
}
)
- Add handler in
handle_call_tool:
elif name == "new_tool_function":
result = self.tools.new_tool_function(
arguments["param1"],
arguments["param2"]
)
📊 What Happens
Event Types Processed:
- DEMAND_SPIKE: Sudden increase in product demand
- DELAY: Supplier delivery delays
- COST_INCREASE: Price changes from suppliers
AI Actions:
- Inventory Transfers: Move stock between warehouses
- Reorder Recommendations: Suggest new orders with quantities
- Stockout Predictions: Forecast when products will run out
- Cost Optimization: Analyze supplier alternatives
🏗️ Architecture
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Flask Web │ │ Custom │ │ Gemini AI │
│ Interface │◄──►│ FastMCP │◄──►│ (or Local │
│ │ │ Server │ │ LLM API) │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│ │
│ │
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ Event Stream │ │ Supply Chain │
│ (CSV Data) │ │ Tools │
└─────────────────┘ └─────────────────┘
📁 Project Structure
FastMCP/
├── data/
│ ├── inventory.csv # Product inventory data
│ └── events.csv # Supply chain events stream
├── templates/
│ └── index.html # Web interface
├── supply_chain_tools.py # Core business logic
├── fastmcp_server.py # Custom FastMCP implementation
├── flask_app.py # Web server and API
├── test_demo.py # Demo script
├── requirements.txt # Python dependencies
└── README.md # This file
🎯 Example Workflow
- Event:
DEMAND_SPIKE for P001 - 40 units - Analysis: AI checks current inventory across warehouses
- Prediction: Identifies potential stockout risk
- Action: Recommends transfer from warehouse B to A
- Execution: Updates inventory and logs the action
Sample Conversation Flow:
Event Stream → MCP Client: "DEMAND_SPIKE: P001, 40 units"
MCP Client → get_inventory_status: {"product_id": "P001"}
MCP Client → predict_stockout: {"product_id": "P001", "warehouse": "warehouse_A"}
MCP Client → calculate_transfer: {"product_id": "P001", "from_warehouse": "warehouse_B", "to_warehouse": "warehouse_A", "quantity": 20}
MCP Client → recommend_reorder: {"product_id": "P001", "quantity": 50}
MCP Client → User: "Transfer 20 units from B to A, reorder 50 units from Supplier X"
🔍 Monitoring
- Terminal Output: Real-time server logs and processing status
- Action Log: All AI recommendations and executed actions
- Status Indicators: Server and event stream status
- Event Progress: Current event being processed
🚀 Key Features
- Real-time Processing: Events processed as they arrive
- Intelligent Recommendations: AI-powered decision making
- Live Updates: Web interface updates in real-time
- Simple Setup: Minimal dependencies and configuration
- Extensible: Easy to add new tools and event types
- Privacy Options: Can use local LLM instead of cloud APIs
🎯 Use Cases
- Supply Chain Optimization: Real-time inventory management
- Demand Forecasting: AI-powered stock predictions
- Cost Optimization: Supplier and pricing analysis
- Risk Management: Stockout prevention and mitigation
🔄 Scenario Modifications
1. Real-Time Supply Chain Optimizer (Streaming Input + Live Agent Correction)
Current Implementation: ✅ Partially Implemented
- ✅ Streaming CSV events
- ✅ Real-time AI responses
- ✅ Basic inventory tools
- ❌ Fast correlation calculator
- ❌ Forecasting tool (ARIMA/exponential smoothing)
- ❌ Live agent correction
What Can Be Added Soon:
# Add to supply_chain_tools.py
def calculate_correlation(self, product1: str, product2: str) -> Dict[str, Any]:
"""Calculate demand correlation between products"""
# Implementation using pandas correlation
def forecast_demand(self, product_id: str, periods: int) -> Dict[str, Any]:
"""Forecast demand using simple exponential smoothing"""
# Implementation using statsmodels
def recommend_reroute(self, from_supplier: str, to_supplier: str) -> Dict[str, Any]:
"""Recommend supply rerouting based on delays/costs"""
# Implementation with cost analysis
Example Conversation:
Event: "SUPPLIER_DELAY: Supplier X, 3 days"
MCP Client: "Analyzing impact on P001, P002, P003..."
Tools Called: [get_inventory_status, calculate_correlation, forecast_demand, recommend_reroute]
Response: "Reroute P001 from Supplier X to Supplier Y. P002 and P003 show 0.8 correlation - adjust P002 orders accordingly."
2. Interactive Survey Analyzer (Multi-Agent & Multi-Tool)
Modification Required:
# New tools in survey_tools.py
def extract_themes(self, responses: List[str]) -> Dict[str, Any]:
"""Extract common themes from survey responses"""
def compute_frequencies(self, data: pd.DataFrame) -> Dict[str, Any]:
"""Compute response frequencies and confidence intervals"""
def generate_summary_report(self, insights: Dict) -> Dict[str, Any]:
"""Generate client-facing summary reports"""
Example Conversation:
User: "Analyze 500 survey responses about Product X"
MCP Client: "Processing responses with multiple agents..."
Tools Called: [extract_themes, compute_frequencies, generate_summary_report]
Response: "Top themes: UI/UX (45%), Performance (32%), Price (23%). 78% satisfaction rate (±3% CI). Report generated."
3. Clinical Triage Assistant (Tool Selection with Tight Latency Loop)
Modification Required:
# New tools in clinical_tools.py
def check_symptoms(self, symptoms: List[str]) -> Dict[str, Any]:
"""Check symptoms against medical database"""
def classify_risk(self, vitals: Dict) -> Dict[str, Any]:
"""Classify patient risk level"""
def score_triage_priority(self, risk: str, symptoms: List) -> Dict[str, Any]:
"""Score triage priority"""
def generate_doctor_note(self, patient_data: Dict) -> Dict[str, Any]:
"""Generate doctor notes"""
Example Conversation:
Patient Data: {"symptoms": ["chest pain", "shortness of breath"], "vitals": {"bp": "140/90"}}
MCP Client: "Analyzing patient data..."
Tools Called: [check_symptoms, classify_risk, score_triage_priority, generate_doctor_note]
Response: "HIGH RISK - Cardiac symptoms detected. Immediate triage required. Doctor note: 'Patient presents with chest pain and elevated BP...'"
4. E-Commerce Pricing Agent (Fast Feedback Loop)
Modification Required:
# New tools in pricing_tools.py
def calculate_optimal_price(self, cost: float, margin: float, demand_factor: float) -> Dict[str, Any]:
"""Calculate optimal price using formula"""
def find_competitor_match(self, product_id: str) -> Dict[str, Any]:
"""Find nearest competitor product"""
def generate_markdown_explanation(self, price_change: Dict) -> Dict[str, Any]:
"""Generate markdown explanation for price changes"""
Example Conversation:
Event: "COMPETITOR_PRICE_CHANGE: Product X, $25.99 → $22.99"
MCP Client: "Analyzing competitive landscape..."
Tools Called: [find_competitor_match, calculate_optimal_price, generate_markdown_explanation]
Response: "Competitor reduced price by 12%. Recommended action: Reduce price to $23.99. Explanation: 'We've adjusted our pricing to remain competitive while maintaining healthy margins...'"
🔧 Development
Adding New Tools
- Add function to
supply_chain_tools.py - Register tool in
fastmcp_server.py - Update system prompts as needed
Adding New Event Types
- Add event to
data/events.csv - Update event processing logic in
fastmcp_server.py - Test with the web interface
Switching to Local LLM
- Set up local-llm-api
- Modify
fastmcp_server.pyto use local API endpoint - Update prompts for local model compatibility
📝 Notes
- This is a demonstration using simulated data
- Inventory changes are saved back to CSV on server stop
- Uses Gemini API free tier (rate limits apply)
- Designed for simplicity and educational purposes
- FastMCP is not open source - this is a custom implementation
- Can be extended with local LLM for data privacy
🤝 Contributing
Feel free to extend this with:
- More sophisticated AI models
- Real database integration
- Additional supply chain tools
- Enhanced web interface features
- Parallel tool execution optimization
- Real-time data streaming
🔗 Related Projects
- local-llm-api: Local LLM API for data privacy
- MCP-RAG: Reference MCP implementation
Ready to optimize your supply chain with AI? Start the server and watch the magic happen! 🚀
This project demonstrates how to build a custom FastMCP-like system for real-time, multi-tool AI orchestration.