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"

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.py to 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

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.

This server cannot be installed

security - not tested

license - not found

quality - not tested

How are these scores calculated?

A custom implementation for real-time supply chain optimization that enables parallel tool calling to provide intelligent inventory management recommendations and actionable insights in response to live supply chain events.

Related MCP Servers

Retail Supply Chain Analysis MCP
jaydrage
-
security
F
license
-
quality
Enables users to upload retail data, analyze trends, optimize inventory, and forecast sales using AI-powered insights, acting as a senior supply chain expert.
Last updated -
JavaScript
MRP Calculator MCP Server
brandon-butterwick
A
security
F
license
A
quality
Provides Material Requirements Planning (MRP) tools for calculating delivery schedules, determining order needs, and performing MRP period calculations based on inventory levels, forecasts, and order constraints.
Last updated -
1
JavaScript
Etherscan MCP Tool
septemhill
A
security
A
license
A
quality
A tool that enables users to interact with blockchain data to retrieve token total supply, chain IDs, and filtered RPC endpoints across different blockchain networks.
Last updated -
6
2
JavaScript
MIT License
chainlink-feeds-mcp
kukapay
-
security
-
license
-
quality
An MCP server that provides real-time access to Chainlink's decentralized on-chain price feeds, optimized for seamless integration into AI agents and autonomous systems.
Last updated -
JavaScript
MIT License

View all related MCP servers