JUHE API Marketplace
ANSH-RIYAL avatar
MCP Server

FastMCP Supply Chain Optimizer

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.

5
GitHub Stars
8/23/2025
Last Updated
No Configuration
Please check the documentation below.

README Documentation

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

  1. Start FastMCP Server: Click "Start FastMCP Server" to initialize the AI agent
  2. Start Event Stream: Click "Start Event Stream" to begin processing supply chain events
  3. Monitor Results: Watch the terminal output and action recommendations in real-time
  4. 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

  1. 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"}
  1. 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"]
    }
)
  1. 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

  1. Event: DEMAND_SPIKE for P001 - 40 units
  2. Analysis: AI checks current inventory across warehouses
  3. Prediction: Identifies potential stockout risk
  4. Action: Recommends transfer from warehouse B to A
  5. 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

  1. Add function to supply_chain_tools.py
  2. Register tool in fastmcp_server.py
  3. Update system prompts as needed

Adding New Event Types

  1. Add event to data/events.csv
  2. Update event processing logic in fastmcp_server.py
  3. Test with the web interface

Switching to Local LLM

  1. Set up local-llm-api
  2. Modify fastmcp_server.py to use local API endpoint
  3. 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


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.

Quick Actions

Key Features

Model Context Protocol
Secure Communication
Real-time Updates
Open Source