Spaces:

MCP-1st-Birthday
/

FalconPrep

Running

App Files Files Community

tsinviking commited on Nov 30, 2025

Commit

ffd6fdf

verified ·

1 Parent(s): bfe3cee

corrected prev commit

Browse files

Files changed (2) hide show

app.py +226 -248
server.py +73 -156

app.py CHANGED Viewed

@@ -1,267 +1,245 @@
 import os
-import math
-from typing import Literal, Optional, Dict, Any, List, Annotated
-from datetime import datetime, timezone
-import requests
-from sgp4.api import Satrec, jday
-from math import sqrt
 from fastmcp import FastMCP
-# from mcp.server.fastmcp import FastMCP
-from langchain_chroma import Chroma
-from langchain_community.embeddings import HuggingFaceEmbeddings
-from langchain_core.vectorstores import VectorStore
-# ==============================================================================
-# 🧠 MCP SERVER INITIALIZATION
-# ==============================================================================
-mcp = FastMCP("FalconPrep", stateless_http=True)
-# ==============================================================================
-# 🧠 CORE CONSTANTS (The Engineering Brain)
-# ==============================================================================
-ENVELOPES: Dict[str, Dict[str, Any]] = {
-    "1U_CubeSat": {"L": 10, "W": 10, "H": 11.35, "max_mass": 2.0},
-    "3U_CubeSat": {"L": 10, "W": 10, "H": 34.05, "max_mass": 5.0},
-    "15_Inch_ESPA": {"L": 61.0, "W": 71.0, "H": 71.0, "max_mass": 220.0},
 }
-PRICING_MODEL: Dict[str, int] = {
-    "base_rate_per_kg": 6500,
-    "min_plate_fee": 300000
 }
-# ==============================================================================
-# 🗄️ RESOURCE: Knowledge Base (VectorStore Connection)
-# ==============================================================================
-DB_PATH = "./falcon_db"
-@mcp.resource("knowledge://rideshare/spacex-manuals-v1")
-def get_knowledge_base_resource() -> Any:
-    print(f"Attempting to load ChromaDB client from {DB_PATH}...")
-    try:
-        embedding_model = HuggingFaceEmbeddings(
-            model_name="all-MiniLM-L6-v2",
-            model_kwargs={'device': 'cpu'}
-        )
-        vectorstore = Chroma(
-            persist_directory=DB_PATH,
-            embedding_function=embedding_model
-        )
-        print("✅ KnowledgeBaseResource loaded successfully.")
-        return vectorstore
-    except Exception as e:
-        print(f"❌ ERROR: Did you run 'python ingest.py'? Error: {e}")
-        return None
-# ==============================================================================
-# 🗣️ PROMPT TEMPLATE
-# ==============================================================================
-@mcp.prompt()
-def launch_readiness_summary_prompt(
-    payload_name: str,
-    fit_check_result: str,
-    hazard_classification_result: Dict[str, Any],
-    cost_estimate: str,
-    required_documents_list: str,
-    timeline_summary: str,
-) -> str:
-    hazard_level = hazard_classification_result.get('level', 'N/A')
-    return f"""
-    You are the **FalconPrep Launch Readiness Assistant**, an expert in SpaceX rideshare compliance.
-    Your task is to synthesize the following tool outputs for payload **'{payload_name}'**.
-    **Guidelines**
-    1. Lead with Fit + Hazard.
-    2. Use bullet points.
-    3. Use professional but friendly compliance language.
-    --- OUTPUTS ---
-    Fit Check:
-    {fit_check_result}
-    Hazard: {hazard_level}
-    Flags: {hazard_classification_result.get('risk_flags', [])}
-    Cost Estimate:
-    {cost_estimate}
-    Required Documents:
-    {required_documents_list}
-    Timeline:
-    {timeline_summary}
-    ---
-    **Produce a final summary now.**
-    """
-# ==============================================================================
-# 🛠️ PAYLOAD TOOLS
-# ==============================================================================
-@mcp.tool()
-def get_launch_requirements(
-    knowledge_base: Annotated[Any, mcp.resource("knowledge://rideshare/spacex-manuals-v1")] = None,
-    payload_type: str = "",
-    orbit: str = "",
-) -> str:
-    query = f"requirements for {payload_type} in {orbit} orbit mechanical electrical communication"
-    try:
-        results = knowledge_base.similarity_search(query, k=3)
-        context = "\n\n".join([doc.page_content for doc in results])
-    except Exception as e:
-        context = f"ERROR during knowledge query: {e}"
-    return f"📄 RAG REQUIREMENTS (Based on manuals):\n{context}"
-@mcp.tool()
-def check_plate_fit(
-    length_cm: float,
-    width_cm: float,
-    height_cm: float,
-    mass_kg: float,
-) -> str:
-    fits, fails = [], []
-    user_dims = sorted([length_cm, width_cm, height_cm])
-    for name, specs in ENVELOPES.items():
-        env_dims = sorted([specs["L"], specs["W"], specs["H"]])
-        mass_ok = mass_kg <= specs["max_mass"]
-        geo_ok = all(u <= e for u, e in zip(user_dims, env_dims))
-        if mass_ok and geo_ok:
-            fits.append(name)
-        else:
-            reasons = []
-            if not mass_ok: reasons.append(f"Overweight (Limit: {specs['max_mass']}kg)")
-            if not geo_ok: reasons.append("Geometry Violation")
-            fails.append(f"{name}: {' + '.join(reasons)}")
-    if fits:
-        return f"✅ FIT SUCCESS: Fits {', '.join(fits)} (Recommend {fits[0]})"
-    return f"❌ FIT FAILURE: No fit. Issues: {chr(10).join(fails)}"
-@mcp.tool()
-def classify_hazard(
-    propellant_type: str,
-    battery_wh: float,
-    pressure_psi: float,
-) -> Dict[str, Any]:
-    classification, flags = "Standard", []
-    if propellant_type.lower() not in ["none", "n/a", "green", "water", "xenon"]:
-        classification, flags = "Hazardous", [f"High-Risk Propellant: {propellant_type}"]
-    if battery_wh > 1000:
-        classification, flags = "Hazardous", flags + [f"Battery > 1kWh ({battery_wh}Wh)"]
-    if pressure_psi > 150:
-        classification, flags = "Hazardous", flags + [f"Pressure > 150 PSI"]
-    if not flags:
-        flags.append("Payload appears standard/benign.")
-    return {
-        "level": classification,
-        "risk_flags": flags,
-        "implication": (
-            "Full multi-phase Safety Review Required"
-            if classification == "Hazardous"
-            else "Standard single-phase review"
-        ),
     }
-@mcp.tool()
-def calculate_launch_cost(mass_kg: float) -> str:
-    mass_cost = mass_kg * PRICING_MODEL["base_rate_per_kg"]
-    final_cost = max(mass_cost, PRICING_MODEL["min_plate_fee"])
-    return f"💰 ESTIMATED COST\nMass Charge: ${mass_cost:,.0f}\nMinimum Plate Fee: ${PRICING_MODEL['min_plate_fee']:,}\n**TOTAL ESTIMATE: ${final_cost:,.0f} USD**"
-@mcp.tool()
-def required_documents(
-    knowledge_base: Annotated[Any, mcp.resource("knowledge://rideshare/spacex-manuals-v1")] = None,
-    payload_type: str = "",
-    hazard_level: Literal["Standard", "Hazardous"] = "Standard",
-) -> str:
-    query = f"required documents for {payload_type} deliverables ICD"
-    try:
-        results = knowledge_base.similarity_search(query, k=3)
-        base_docs = "\n\n".join([doc.page_content for doc in results])
-    except Exception as e:
-        base_docs = f"ERROR during knowledge query: {e}"
-    extra = ""
-    if hazard_level == "Hazardous":
-        extra = "\n⚠️ EXTRA HAZARDOUS DOCUMENTS:\n* MSDS for all fluids\n* Burst Test Certificate\n* Propellant Handling Plan\n* Full Safety Review Package (SRDP)"
-    return f"📑 REQUIRED DOCUMENTS ({payload_type}, {hazard_level})\n{extra}\n---\nSTANDARD (from manual):\n{base_docs}"
-@mcp.tool()
-def timeline_check(
-    knowledge_base: Annotated[Any, mcp.resource("knowledge://rideshare/spacex-manuals-v1")] = None,
-    hazard_level: Literal["Standard", "Hazardous"] = "Standard",
-) -> str:
-    query = "launch campaign schedule L-minus integration deadlines"
-    try:
-        results = knowledge_base.similarity_search(query, k=3)
-        base_timeline = "\n\n".join([doc.page_content for doc in results])
-    except Exception as e:
-        base_timeline = f"ERROR during knowledge query: {e}"
-    safety = "✅ Standard Review ~L-4 Months" if hazard_level == "Standard" else "\n🛑 HAZARDOUS EARLY REVIEWS:\n* L-12m: Phase 0\n* L-9m: Phase 1\n* L-6m: Phase 2\n* L-3m: Phase 3"
-    return f"🕒 TIMELINE ({hazard_level})\n{safety}\n---\nSTANDARD MILESTONES:\n{base_timeline}"
-# ==============================================================================
-# 🛰️ ORBITAL TOOLS
-# ==============================================================================
-@mcp.tool()
-def fetch_gp_data_tool(
-    query_type: str = "CATNR",
-    query_value: str = "",
-    format: Literal["TLE", "JSON", "JSON-PRETTY", "CSV", "XML", "KVN"] = "JSON"
-) -> Dict[str, Any]:
-    base_url = "https://celestrak.org/NORAD/elements/gp.php"
-    params = {"FORMAT": format, query_type.upper(): query_value}
-    try:
-        resp = requests.get(base_url, params=params, timeout=10)
-        resp.raise_for_status()
-        if "JSON" in format:
-            return resp.json()
-        else:
-            return {"raw": resp.text}
-    except Exception as e:
-        return {"error": str(e)}
-@mcp.tool()
-def propagate_orbit_tool(
-    tle_lines: List[str],
-    target_time: Optional[datetime] = None
-) -> Dict[str, Any]:
-    try:
-        sat = Satrec.twoline2rv(tle_lines[0], tle_lines[1])
-        target_time = target_time or datetime.now(timezone.utc)
-        jd, fr = jday(target_time.year, target_time.month, target_time.day,
-                      target_time.hour, target_time.minute, target_time.second + target_time.microsecond*1e-6)
-        e, r, v = sat.sgp4(jd, fr)
-        if e != 0:
-            raise RuntimeError(f"SGP4 error code {e}")
-        return {"position_km": tuple(r), "velocity_kms": tuple(v), "timestamp": target_time.isoformat()}
-    except Exception as e:
-        return {"error": str(e)}
-@mcp.tool()
-def collision_check_tool(
-    sat1_tle: List[str],
-    sat2_tle: List[str],
-    threshold_km: float = 5.0,
-    target_time: Optional[datetime] = None
-) -> Dict[str, Any]:
-    try:
-        pos1 = propagate_orbit_tool(sat1_tle, target_time)["position_km"]
-        pos2 = propagate_orbit_tool(sat2_tle, target_time)["position_km"]
-        distance = sqrt(sum((a - b) ** 2 for a, b in zip(pos1, pos2)))
-        warning = distance <= threshold_km
-        return {"distance_km": distance, "collision_warning": warning, "threshold_km": threshold_km, "timestamp": (target_time or datetime.now(timezone.utc)).isoformat()}
-    except Exception as e:
-        return {"error": str(e)}
-# ==============================================================================
-# 🏁 RUN MCP SERVER
-# ==============================================================================
 if __name__ == "__main__":
-    mcp.run()

 import os
 from fastmcp import FastMCP
+from fastapi.responses import HTMLResponse
+from fastapi.middleware.cors import CORSMiddleware
+from server import mcp
+# app = mcp.app
+# app.add_middleware(
+#     CORSMiddleware,
+#     allow_origins=["*"],
+#     allow_credentials=True,
+#     allow_methods=["*"],
+#     allow_headers=["*"],
+# )
+# -----------------------------
+# Landing Page (Optional UI)
+# -----------------------------
+@mcp.custom_route("/", methods=["GET"])
+async def index(_):
+    return HTMLResponse("""
+<!DOCTYPE html>
+<html lang="en">
+<head>
+<meta charset="UTF-8">
+<meta name="viewport" content="width=device-width, initial-scale=1.0">
+<title>FalconPrep MCP Server</title>
+<style>
+    :root {
+        --accent: #22d3ee;
+        --bg: #020b14;
+        --panel: rgba(0,0,0,0.45);
+        --border: #1e293b;
+        --font: "Roboto Mono", monospace;
+        --text-dim: #9ac7e0;
+    }
+    body {
+        margin: 0;
+        background: var(--bg);
+        font-family: var(--font);
+        color: white;
+        overflow-x: hidden;
+        overflow-y: auto;
+        min-height: 100vh;
+        position: relative;
+    }
+    #starfield {
+        position: fixed;
+        inset: 0;
+        width: 100vw;
+        height: 100vh;
+        z-index: -1;
+        display: block;
+    }
+    .container {
+        width: 85vw;
+        max-width: 900px;
+        margin: 0 auto;
+        padding-top: 10vh;
+        padding-bottom: 10vh;
+        display: flex;
+        flex-direction: column;
+        gap: 2.5vh;
+    }
+    h1 {
+        font-size: 3.2rem;
+        text-transform: uppercase;
+        letter-spacing: 0.18rem;
+        margin: 0 0 0.5rem 0;
+        color: var(--accent);
+        font-weight: 900;
+        text-align: center;
+    }
+    .subtitle {
+        opacity: 0.75;
+        letter-spacing: 0.12rem;
+        text-align: center;
+        margin-bottom: 2vh;
+    }
+    .panel {
+        background: var(--panel);
+        border: 1px solid var(--border);
+        border-radius: 8px;
+        padding: 1.4rem;
+        box-shadow: 0px 0px 15px rgba(34,211,238,0.08);
+        backdrop-filter: blur(4px);
+    }
+    .panel h2 {
+        margin-top: 0;
+        margin-bottom: 0.8rem;
+        font-size: 1.25rem;
+        color: var(--accent);
+        font-weight: 700;
+        text-transform: uppercase;
+        letter-spacing: 0.12rem;
+    }
+    p {
+        font-size: 1rem;
+        line-height: 1.55;
+        color: var(--text-dim);
+    }
+    ul {
+        padding-left: 1.2rem;
+        margin: 0;
+    }
+    li {
+        color: var(--text-dim);
+        margin-bottom: 0.5rem;
+        font-size: 1rem;
+    }
+    code {
+        color: var(--accent);
+    }
+</style>
+</head>
+<body>
+<canvas id="starfield"></canvas>
+<div class="container">
+    <h1>FalconPrep</h1>
+    <p class="subtitle">AI Launch Integration Assistant</p>
+    <div class="panel"><h2>Problem</h2>
+        <p>Launch teams must navigate 200–400+ pages of SpaceX rideshare manuals, unclear pricing, and slow email back-and-forth.</p>
+    </div>
+    <div class="panel"><h2>Pain Points</h2>
+        <ul>
+            <li>Complex requirements & engineering jargon</li>
+            <li>Slow cost estimation</li>
+            <li>Unclear safety & propellant rules</li>
+            <li>Scattered planning tools (PDFs, Excel, emails)</li>
+        </ul>
+    </div>
+    <div class="panel"><h2>Solution</h2>
+        <p>FalconPrep turns the documentation into structured MCP tools so AI agents can compute requirements, costs, hazards, and timelines.</p>
+    </div>
+    <div class="panel"><h2>Core Tools</h2>
+        <ul>
+            <li>check_plate_fit</li>
+            <li>classify_hazard</li>
+            <li>estimate_cost</li>
+            <li>lookup_standard</li>
+            <li>generate_report</li>
+        </ul>
+    </div>
+    <div class="panel"><h2>How to Use</h2>
+        <ul>
+            <li>Connect with an MCP client</li>
+            <li>SSE endpoint: <code>/sse</code></li>
+            <li>Supports all tool calls via JSON</li>
+        </ul>
+    </div>
+</div>
+<script>
+/* starfield js animation preserved */
+const canvas = document.getElementById("starfield");
+const ctx = canvas.getContext("2d");
+let stars = [];
+function resizeCanvas() {
+    canvas.width = window.innerWidth;
+    canvas.height = window.innerHeight;
 }
+function initStars() {
+    resizeCanvas();
+    stars = Array.from({ length: 450 }, () => ({
+        x: Math.random() * canvas.width,
+        y: Math.random() * canvas.height,
+        speed: Math.random() * 0.7 + 0.2
+    }));
 }
+let lastScrollY = 0;
+let scrollSpeed = 0;
+window.addEventListener("scroll", () => {
+    const current = window.scrollY;
+    scrollSpeed = current - lastScrollY;
+    lastScrollY = current;
+});
+let smoothScroll = 0;
+function animateStars() {
+    smoothScroll += (scrollSpeed - smoothScroll) * 0.05;
+    ctx.fillStyle = "#020b14";
+    ctx.fillRect(0, 0, canvas.width, canvas.height);
+    ctx.fillStyle = "white";
+    for (const s of stars) {
+        ctx.fillRect(s.x, s.y, 1.5, 1.5);
+        const scrollBoost = smoothScroll * 0.05;
+        s.y += s.speed + scrollBoost;
+        if (s.y > canvas.height) s.y -= canvas.height;
+        if (s.y < 0) s.y += canvas.height;
     }
+    requestAnimationFrame(animateStars);
+}
+window.addEventListener("resize", resizeCanvas);
+initStars();
+animateStars();
+</script>
+</body>
+</html>
+""")
+# -----------------------------
+# Run FastMCP (with SSE)
+# -----------------------------
+HF_SPACE_PORT = int(os.getenv("PORT", 7860))
+HF_SPACE_HOST = os.getenv("HOST", "0.0.0.0")
 if __name__ == "__main__":
+    print("🚀 FalconPrep MCP Server starting...")
+    if hasattr(mcp, '_app'):
+        mcp._app.add_middleware(
+            CORSMiddleware,
+            allow_origins=["*"],
+            allow_credentials=True,
+            allow_methods=["*"],
+            allow_headers=["*"],
+        )
+    mcp.run(
+        transport="sse",
+        host=HF_SPACE_HOST,
+        port=HF_SPACE_PORT,
+    )

server.py CHANGED Viewed

@@ -1,6 +1,10 @@
 import os
 import math
 from typing import Literal, Optional, Dict, Any, List, Annotated
 from fastmcp import FastMCP
 # from mcp.server.fastmcp import FastMCP
@@ -8,7 +12,10 @@ from langchain_chroma import Chroma
 from langchain_community.embeddings import HuggingFaceEmbeddings
 from langchain_core.vectorstores import VectorStore
-# Initialize MCP Server
 mcp = FastMCP("FalconPrep", stateless_http=True)
 # ==============================================================================
@@ -34,16 +41,6 @@ DB_PATH = "./falcon_db"
 @mcp.resource("knowledge://rideshare/spacex-manuals-v1")
 def get_knowledge_base_resource() -> Any:
-    """
-    Initializes and returns the query client for the persistent ChromaDB vector store.
-    This resource loads the SpaceX Payload User Guides and uses a local
-    HuggingFace embedding model for zero-cost operation. The return type is
-    Any to prevent Pydantic serialization errors during FastMCP startup.
-    Returns:
-        Any (Chroma VectorStore instance): The connected vector store object.
-    """
     print(f"Attempting to load ChromaDB client from {DB_PATH}...")
     try:
         embedding_model = HuggingFaceEmbeddings(
@@ -54,7 +51,6 @@ def get_knowledge_base_resource() -> Any:
             persist_directory=DB_PATH,
             embedding_function=embedding_model
         )
         print("✅ KnowledgeBaseResource loaded successfully.")
         return vectorstore
     except Exception as e:
@@ -74,21 +70,6 @@ def launch_readiness_summary_prompt(
     required_documents_list: str,
     timeline_summary: str,
 ) -> str:
-    """
-    Instructs the AI Agent on how to synthesize the structured tool outputs
-    into a comprehensive, final launch readiness report for the user.
-    Args:
-        payload_name (str): The name of the satellite/payload.
-        fit_check_result (str): Output from the check_plate_fit tool.
-        hazard_classification_result (Dict[str, Any]): Output from the classify_hazard tool.
-        cost_estimate (str): Output from the calculate_launch_cost tool.
-        required_documents_list (str): Output from the required_documents tool.
-        timeline_summary (str): Output from the timeline_check tool.
-    Returns:
-        str: The full system prompt instructing the LLM on synthesis.
-    """
     hazard_level = hazard_classification_result.get('level', 'N/A')
     return f"""
     You are the **FalconPrep Launch Readiness Assistant**, an expert in SpaceX rideshare compliance.
@@ -120,7 +101,7 @@ def launch_readiness_summary_prompt(
     """
 # ==============================================================================
-# 🛠️ TOOLS
 # ==============================================================================
 @mcp.tool()
@@ -129,30 +110,13 @@ def get_launch_requirements(
     payload_type: str = "",
     orbit: str = "",
 ) -> str:
-    """
-    RAG-Based: Retrieves qualitative launch requirements from the ingested SpaceX payload manuals.
-    Queries the vector store for detailed technical and integration requirements
-    based on the payload type and target orbit.
-    Args:
-        knowledge_base (Any): The injected Chroma VectorStore client.
-        payload_type (str): The type of payload (e.g., 'CubeSat', 'Cake Topper').
-        orbit (str): The intended orbit (e.g., 'SSO', 'LEO').
-    Returns:
-        str: A string containing the relevant context found in the manuals.
-    """
     query = f"requirements for {payload_type} in {orbit} orbit mechanical electrical communication"
     try:
         results = knowledge_base.similarity_search(query, k=3)
         context = "\n\n".join([doc.page_content for doc in results])
     except Exception as e:
         context = f"ERROR during knowledge query: {e}"
-    return f"""
-    📄 RAG REQUIREMENTS (Based on manuals):
-    {context}
-    """
 @mcp.tool()
 def check_plate_fit(
@@ -161,29 +125,12 @@ def check_plate_fit(
     height_cm: float,
     mass_kg: float,
 ) -> str:
-    """
-    Logic Tool: Checks if a payload's bounding box and mass fit standard rideshare interfaces.
-    Compares the user-provided dimensions and mass against predefined standards
-    (1U, 3U, ESPA-class).
-    Args:
-        length_cm (float): Payload length in centimeters.
-        width_cm (float): Payload width in centimeters.
-        height_cm (float): Payload height in centimeters.
-        mass_kg (float): Payload mass in kilograms.
-    Returns:
-        str: A string indicating success and compatible interfaces, or a failure detail.
-    """
     fits, fails = [], []
     user_dims = sorted([length_cm, width_cm, height_cm])
     for name, specs in ENVELOPES.items():
         env_dims = sorted([specs["L"], specs["W"], specs["H"]])
         mass_ok = mass_kg <= specs["max_mass"]
         geo_ok = all(u <= e for u, e in zip(user_dims, env_dims))
         if mass_ok and geo_ok:
             fits.append(name)
         else:
@@ -191,7 +138,6 @@ def check_plate_fit(
             if not mass_ok: reasons.append(f"Overweight (Limit: {specs['max_mass']}kg)")
             if not geo_ok: reasons.append("Geometry Violation")
             fails.append(f"{name}: {' + '.join(reasons)}")
     if fits:
         return f"✅ FIT SUCCESS: Fits {', '.join(fits)} (Recommend {fits[0]})"
     return f"❌ FIT FAILURE: No fit. Issues: {chr(10).join(fails)}"
@@ -202,32 +148,15 @@ def classify_hazard(
     battery_wh: float,
     pressure_psi: float,
 ) -> Dict[str, Any]:
-    """
-    Logic Tool: Determines the payload's Hazard Classification (Standard/Hazardous).
-    Classification is based on critical thresholds for propulsion, battery capacity
-    (>1 kWh or 1000 Wh), and pressure vessels (>150 PSI).
-    Args:
-        propellant_type (str): Type of propellant used (e.g., Hydrazine, Xenon, None).
-        battery_wh (float): Total energy capacity of batteries in Watt-hours.
-        pressure_psi (float): Max operating pressure of any vessel in PSI.
-    Returns:
-        Dict[str, Any]: Classification level, risk flags, and implication.
-    """
     classification, flags = "Standard", []
     if propellant_type.lower() not in ["none", "n/a", "green", "water", "xenon"]:
         classification, flags = "Hazardous", [f"High-Risk Propellant: {propellant_type}"]
     if battery_wh > 1000:
         classification, flags = "Hazardous", flags + [f"Battery > 1kWh ({battery_wh}Wh)"]
     if pressure_psi > 150:
         classification, flags = "Hazardous", flags + [f"Pressure > 150 PSI"]
     if not flags:
         flags.append("Payload appears standard/benign.")
     return {
         "level": classification,
         "risk_flags": flags,
@@ -240,26 +169,9 @@ def classify_hazard(
 @mcp.tool()
 def calculate_launch_cost(mass_kg: float) -> str:
-    """
-    Logic Tool: Calculates the estimated launch cost using a linear cost model.
-    Applies the base rate per kg, but enforces a minimum plate fee defined in
-    the CORE CONSTANTS.
-    Args:
-        mass_kg (float): Payload mass in kilograms.
-    Returns:
-        str: A formatted string detailing the mass cost, minimum fee, and total estimate.
-    """
     mass_cost = mass_kg * PRICING_MODEL["base_rate_per_kg"]
     final_cost = max(mass_cost, PRICING_MODEL["min_plate_fee"])
-    return f"""
-    💰 ESTIMATED COST
-    Mass Charge: ${mass_cost:,.0f}
-    Minimum Plate Fee: ${PRICING_MODEL['min_plate_fee']:,}
-    **TOTAL ESTIMATE: ${final_cost:,.0f} USD**
-    """
 @mcp.tool()
 def required_documents(
@@ -267,84 +179,89 @@ def required_documents(
     payload_type: str = "",
     hazard_level: Literal["Standard", "Hazardous"] = "Standard",
 ) -> str:
-    """
-    RAG & Logic Hybrid: Generates a document checklist, dynamically adding
-    specific safety documents if the payload is classified as Hazardous.
-    Args:
-        knowledge_base (Any): The injected Chroma VectorStore client.
-        payload_type (str): The payload type for RAG querying.
-        hazard_level (Literal["Standard", "Hazardous"]): The classification from classify_hazard.
-    Returns:
-        str: The combined list of standard and hazard-specific documentation.
-    """
     query = f"required documents for {payload_type} deliverables ICD"
     try:
         results = knowledge_base.similarity_search(query, k=3)
         base_docs = "\n\n".join([doc.page_content for doc in results])
     except Exception as e:
         base_docs = f"ERROR during knowledge query: {e}"
     extra = ""
     if hazard_level == "Hazardous":
-        extra = """
-        ⚠️ EXTRA HAZARDOUS DOCUMENTS:
-        * MSDS for all fluids
-        * Burst Test Certificate
-        * Propellant Handling Plan
-        * Full Safety Review Package (SRDP)
-        """
-    return f"""
-    📑 REQUIRED DOCUMENTS ({payload_type}, {hazard_level})
-    {extra}
-    ---
-    STANDARD (from manual):
-    {base_docs}
-    """
 @mcp.tool()
 def timeline_check(
     knowledge_base: Annotated[Any, mcp.resource("knowledge://rideshare/spacex-manuals-v1")] = None,
     hazard_level: Literal["Standard", "Hazardous"] = "Standard",
 ) -> str:
-    """
-    RAG & Logic Hybrid: Generates a timeline summary, enforcing stricter safety
-    review deadlines for Hazardous payloads based on internal logic.
-    Args:
-        knowledge_base (Any): The injected Chroma VectorStore client.
-        hazard_level (Literal["Standard", "Hazardous"]): The classification from classify_hazard.
-    Returns:
-        str: The summary including safety milestones and standard integration dates.
-    """
     query = "launch campaign schedule L-minus integration deadlines"
     try:
         results = knowledge_base.similarity_search(query, k=3)
         base_timeline = "\n\n".join([doc.page_content for doc in results])
     except Exception as e:
         base_timeline = f"ERROR during knowledge query: {e}"
-    if hazard_level == "Hazardous":
-        safety = """
-        🛑 HAZARDOUS EARLY REVIEWS:
-        * L-12m: Phase 0
-        * L-9m: Phase 1
-        * L-6m: Phase 2
-        * L-3m: Phase 3
-        """
-    else:
-        safety = "✅ Standard Review ~L-4 Months"
-    return f"""
-    🕒 TIMELINE ({hazard_level})
-    {safety}
-    ---
-    STANDARD MILESTONES:
-    {base_timeline}
-    """
 if __name__ == "__main__":
-    mcp.run()

 import os
 import math
 from typing import Literal, Optional, Dict, Any, List, Annotated
+from datetime import datetime, timezone
+import requests
+from sgp4.api import Satrec, jday
+from math import sqrt
 from fastmcp import FastMCP
 # from mcp.server.fastmcp import FastMCP
 from langchain_community.embeddings import HuggingFaceEmbeddings
 from langchain_core.vectorstores import VectorStore
+# ==============================================================================
+# 🧠 MCP SERVER INITIALIZATION
+# ==============================================================================
 mcp = FastMCP("FalconPrep", stateless_http=True)
 # ==============================================================================
 @mcp.resource("knowledge://rideshare/spacex-manuals-v1")
 def get_knowledge_base_resource() -> Any:
     print(f"Attempting to load ChromaDB client from {DB_PATH}...")
     try:
         embedding_model = HuggingFaceEmbeddings(
             persist_directory=DB_PATH,
             embedding_function=embedding_model
         )
         print("✅ KnowledgeBaseResource loaded successfully.")
         return vectorstore
     except Exception as e:
     required_documents_list: str,
     timeline_summary: str,
 ) -> str:
     hazard_level = hazard_classification_result.get('level', 'N/A')
     return f"""
     You are the **FalconPrep Launch Readiness Assistant**, an expert in SpaceX rideshare compliance.
     """
 # ==============================================================================
+# 🛠️ PAYLOAD TOOLS
 # ==============================================================================
 @mcp.tool()
     payload_type: str = "",
     orbit: str = "",
 ) -> str:
     query = f"requirements for {payload_type} in {orbit} orbit mechanical electrical communication"
     try:
         results = knowledge_base.similarity_search(query, k=3)
         context = "\n\n".join([doc.page_content for doc in results])
     except Exception as e:
         context = f"ERROR during knowledge query: {e}"
+    return f"📄 RAG REQUIREMENTS (Based on manuals):\n{context}"
 @mcp.tool()
 def check_plate_fit(
     height_cm: float,
     mass_kg: float,
 ) -> str:
     fits, fails = [], []
     user_dims = sorted([length_cm, width_cm, height_cm])
     for name, specs in ENVELOPES.items():
         env_dims = sorted([specs["L"], specs["W"], specs["H"]])
         mass_ok = mass_kg <= specs["max_mass"]
         geo_ok = all(u <= e for u, e in zip(user_dims, env_dims))
         if mass_ok and geo_ok:
             fits.append(name)
         else:
             if not mass_ok: reasons.append(f"Overweight (Limit: {specs['max_mass']}kg)")
             if not geo_ok: reasons.append("Geometry Violation")
             fails.append(f"{name}: {' + '.join(reasons)}")
     if fits:
         return f"✅ FIT SUCCESS: Fits {', '.join(fits)} (Recommend {fits[0]})"
     return f"❌ FIT FAILURE: No fit. Issues: {chr(10).join(fails)}"
     battery_wh: float,
     pressure_psi: float,
 ) -> Dict[str, Any]:
     classification, flags = "Standard", []
     if propellant_type.lower() not in ["none", "n/a", "green", "water", "xenon"]:
         classification, flags = "Hazardous", [f"High-Risk Propellant: {propellant_type}"]
     if battery_wh > 1000:
         classification, flags = "Hazardous", flags + [f"Battery > 1kWh ({battery_wh}Wh)"]
     if pressure_psi > 150:
         classification, flags = "Hazardous", flags + [f"Pressure > 150 PSI"]
     if not flags:
         flags.append("Payload appears standard/benign.")
     return {
         "level": classification,
         "risk_flags": flags,
 @mcp.tool()
 def calculate_launch_cost(mass_kg: float) -> str:
     mass_cost = mass_kg * PRICING_MODEL["base_rate_per_kg"]
     final_cost = max(mass_cost, PRICING_MODEL["min_plate_fee"])
+    return f"💰 ESTIMATED COST\nMass Charge: ${mass_cost:,.0f}\nMinimum Plate Fee: ${PRICING_MODEL['min_plate_fee']:,}\n**TOTAL ESTIMATE: ${final_cost:,.0f} USD**"
 @mcp.tool()
 def required_documents(
     payload_type: str = "",
     hazard_level: Literal["Standard", "Hazardous"] = "Standard",
 ) -> str:
     query = f"required documents for {payload_type} deliverables ICD"
     try:
         results = knowledge_base.similarity_search(query, k=3)
         base_docs = "\n\n".join([doc.page_content for doc in results])
     except Exception as e:
         base_docs = f"ERROR during knowledge query: {e}"
     extra = ""
     if hazard_level == "Hazardous":
+        extra = "\n⚠️ EXTRA HAZARDOUS DOCUMENTS:\n* MSDS for all fluids\n* Burst Test Certificate\n* Propellant Handling Plan\n* Full Safety Review Package (SRDP)"
+    return f"📑 REQUIRED DOCUMENTS ({payload_type}, {hazard_level})\n{extra}\n---\nSTANDARD (from manual):\n{base_docs}"
 @mcp.tool()
 def timeline_check(
     knowledge_base: Annotated[Any, mcp.resource("knowledge://rideshare/spacex-manuals-v1")] = None,
     hazard_level: Literal["Standard", "Hazardous"] = "Standard",
 ) -> str:
     query = "launch campaign schedule L-minus integration deadlines"
     try:
         results = knowledge_base.similarity_search(query, k=3)
         base_timeline = "\n\n".join([doc.page_content for doc in results])
     except Exception as e:
         base_timeline = f"ERROR during knowledge query: {e}"
+    safety = "✅ Standard Review ~L-4 Months" if hazard_level == "Standard" else "\n🛑 HAZARDOUS EARLY REVIEWS:\n* L-12m: Phase 0\n* L-9m: Phase 1\n* L-6m: Phase 2\n* L-3m: Phase 3"
+    return f"🕒 TIMELINE ({hazard_level})\n{safety}\n---\nSTANDARD MILESTONES:\n{base_timeline}"
+# ==============================================================================
+# 🛰️ ORBITAL TOOLS
+# ==============================================================================
+@mcp.tool()
+def fetch_gp_data_tool(
+    query_type: str = "CATNR",
+    query_value: str = "",
+    format: Literal["TLE", "JSON", "JSON-PRETTY", "CSV", "XML", "KVN"] = "JSON"
+) -> Dict[str, Any]:
+    base_url = "https://celestrak.org/NORAD/elements/gp.php"
+    params = {"FORMAT": format, query_type.upper(): query_value}
+    try:
+        resp = requests.get(base_url, params=params, timeout=10)
+        resp.raise_for_status()
+        if "JSON" in format:
+            return resp.json()
+        else:
+            return {"raw": resp.text}
+    except Exception as e:
+        return {"error": str(e)}
+@mcp.tool()
+def propagate_orbit_tool(
+    tle_lines: List[str],
+    target_time: Optional[datetime] = None
+) -> Dict[str, Any]:
+    try:
+        sat = Satrec.twoline2rv(tle_lines[0], tle_lines[1])
+        target_time = target_time or datetime.now(timezone.utc)
+        jd, fr = jday(target_time.year, target_time.month, target_time.day,
+                      target_time.hour, target_time.minute, target_time.second + target_time.microsecond*1e-6)
+        e, r, v = sat.sgp4(jd, fr)
+        if e != 0:
+            raise RuntimeError(f"SGP4 error code {e}")
+        return {"position_km": tuple(r), "velocity_kms": tuple(v), "timestamp": target_time.isoformat()}
+    except Exception as e:
+        return {"error": str(e)}
+@mcp.tool()
+def collision_check_tool(
+    sat1_tle: List[str],
+    sat2_tle: List[str],
+    threshold_km: float = 5.0,
+    target_time: Optional[datetime] = None
+) -> Dict[str, Any]:
+    try:
+        pos1 = propagate_orbit_tool(sat1_tle, target_time)["position_km"]
+        pos2 = propagate_orbit_tool(sat2_tle, target_time)["position_km"]
+        distance = sqrt(sum((a - b) ** 2 for a, b in zip(pos1, pos2)))
+        warning = distance <= threshold_km
+        return {"distance_km": distance, "collision_warning": warning, "threshold_km": threshold_km, "timestamp": (target_time or datetime.now(timezone.utc)).isoformat()}
+    except Exception as e:
+        return {"error": str(e)}
+# ==============================================================================
+# 🏁 RUN MCP SERVER
+# ==============================================================================
 if __name__ == "__main__":
+    mcp.run()