Update app.py

app.py

@@ -1,7 +1,5 @@
-# app.py — MCP server using an open-source local LLM (transformers) or a rule-based fallback
-# - Uses FastMCP for tools
-# - Gradio ChatInterface for UI
-# - process_document accepts local path and transforms it to a file:// URL in the tool call
+# app.py — MCP POC using local Hugging Face model (flan-t5 or other) or rule-based fallback.
+# Place this file next to config.py. Do NOT store secrets here.
 
 from mcp.server.fastmcp import FastMCP
 from typing import Optional, List, Tuple, Any, Dict
@@ -14,21 +12,14 @@ import traceback
 import inspect
 import re
 
-# Optional imports for local model
+# Optional transformers imports — load only if available
+TRANSFORMERS_AVAILABLE = False
 try:
-    from transformers import pipeline, AutoModelForCausalLM, AutoTokenizer
+    from transformers import pipeline, AutoTokenizer, AutoModelForSeq2SeqLM, AutoModelForCausalLM
     TRANSFORMERS_AVAILABLE = True
 except Exception:
     TRANSFORMERS_AVAILABLE = False
 
-# Optional embeddings for light retrieval if desired
-try:
-    from sentence_transformers import SentenceTransformer
-    import numpy as np
-    SENTEVAL_AVAILABLE = True
-except Exception:
-    SENTEVAL_AVAILABLE = False
-
 # ----------------------------
 # Load config
 # ----------------------------
@@ -38,8 +29,8 @@ try:
         CLIENT_SECRET,
         REFRESH_TOKEN,
         API_BASE,
-        LOCAL_MODEL,        # e.g. "tiiuae/falcon-7b-instruct" if you have it locally
-        LOCAL_TOKENIZER,
+        LOCAL_MODEL,      # e.g. "google/flan-t5-base" or None
+        LOCAL_TOKENIZER   # optional: tokenizer name if different
     )
 except Exception:
     raise SystemExit(
@@ -52,17 +43,21 @@ except Exception:
 mcp = FastMCP("ZohoCRMAgent")
 
 # ----------------------------
-# Analytics (simple)
+# Analytics / KPI logging (simple local JSON file)
 # ----------------------------
 ANALYTICS_PATH = "mcp_analytics.json"
 
 def _init_analytics():
     if not os.path.exists(ANALYTICS_PATH):
-        base = {"tool_calls": {}, "llm_calls": 0, "last_llm_confidence": None, "created_at": time.time()}
+        base = {
+            "tool_calls": {},
+            "llm_calls": 0,
+            "last_llm_confidence": None,
+            "created_at": time.time()
+        }
         with open(ANALYTICS_PATH, "w") as f:
             json.dump(base, f, indent=2)
 
-
 def _log_tool_call(tool_name: str, success: bool = True):
     try:
         with open(ANALYTICS_PATH, "r") as f:
@@ -78,7 +73,6 @@ def _log_tool_call(tool_name: str, success: bool = True):
     with open(ANALYTICS_PATH, "w") as f:
         json.dump(data, f, indent=2)
 
-
 def _log_llm_call(confidence: Optional[float] = None):
     try:
         with open(ANALYTICS_PATH, "r") as f:
@@ -94,45 +88,71 @@ def _log_llm_call(confidence: Optional[float] = None):
 _init_analytics()
 
 # ----------------------------
-# Local LLM: attempt to load transformers pipeline
+# Local LLM pipeline initialization
 # ----------------------------
 LLM_PIPELINE = None
 TOKENIZER = None
 
 def init_local_model():
+    """
+    Initialize local HF model pipeline depending on LOCAL_MODEL.
+    Supports seq2seq (flan/t5) and causal models.
+    If transformers is unavailable or LOCAL_MODEL is None, leaves LLM_PIPELINE as None.
+    """
     global LLM_PIPELINE, TOKENIZER
-    if not TRANSFORMERS_AVAILABLE or not LOCAL_MODEL:
-        print("Local transformers not available or LOCAL_MODEL not set — falling back to rule-based responder.")
+
+    if not LOCAL_MODEL:
+        print("LOCAL_MODEL is None — using rule-based fallback.")
+        LLM_PIPELINE = None
+        return
+
+    if not TRANSFORMERS_AVAILABLE:
+        print("transformers not installed — using rule-based fallback.")
+        LLM_PIPELINE = None
         return
+
     try:
-        # If a specific tokenizer name was provided use it, otherwise use model name
         tokenizer_name = LOCAL_TOKENIZER or LOCAL_MODEL
-        TOKENIZER = AutoTokenizer.from_pretrained(tokenizer_name)
-        model = AutoModelForCausalLM.from_pretrained(LOCAL_MODEL, device_map="auto")
-        LLM_PIPELINE = pipeline("text-generation", model=model, tokenizer=TOKENIZER)
-        print(f"Loaded local model: {LOCAL_MODEL}")
+
+        # Detect seq2seq family (T5/Flan)
+        if any(x in LOCAL_MODEL.lower() for x in ["flan", "t5", "seq2seq"]):
+            TOKENIZER = AutoTokenizer.from_pretrained(tokenizer_name, use_fast=True)
+            model = AutoModelForSeq2SeqLM.from_pretrained(LOCAL_MODEL)
+            LLM_PIPELINE = pipeline("text2text-generation", model=model, tokenizer=TOKENIZER)
+            print(f"Loaded seq2seq model pipeline for {LOCAL_MODEL}")
+        else:
+            # causal model path
+            TOKENIZER = AutoTokenizer.from_pretrained(tokenizer_name, use_fast=True)
+            model = AutoModelForCausalLM.from_pretrained(LOCAL_MODEL)
+            LLM_PIPELINE = pipeline("text-generation", model=model, tokenizer=TOKENIZER)
+            print(f"Loaded causal model pipeline for {LOCAL_MODEL}")
+
     except Exception as e:
         print("Failed to load local model:", e)
+        traceback.print_exc()
         LLM_PIPELINE = None
 
+# Try to init model at startup (may be slow)
 init_local_model()
 
 # ----------------------------
-# Simple rule-based responder fallback
+# Rule-based fallback responder
 # ----------------------------
 def rule_based_response(message: str) -> str:
-    msg = message.lower()
+    msg = (message or "").strip().lower()
     if msg.startswith("create record") or msg.startswith("create contact"):
-        return "To create a record, say: create_record MODULENAME {\\\"Field\\\": \\\"value\\\"}"
+        return "To create a record, use the command: create_record MODULE_NAME {\"Field\": \"value\"}"
+    if msg.startswith("create_invoice"):
+        return "To create invoice: create_invoice {\"customer_id\": \"...\", \"line_items\": [...]} (JSON)"
     if msg.startswith("help") or msg.startswith("what can you do"):
-        return "I can create/update/delete records in Zoho (create_record, update_record, delete_record), or process local files by pasting their path (/mnt/data/...)."
-    return "(fallback) I don't have a local model loaded. Use a supported local LLM or call create_record directly."
+        return "I can create/update/delete records in Zoho (create_record/update_record/delete_record) or process local files by pasting their path (/mnt/data/...)."
+    return "(fallback) No local LLM loaded. Use explicit commands like `create_record` or paste a /mnt/data/ path."
 
 # ----------------------------
-# Zoho token & MCP tools — same patterns as before
+# Zoho token refresh & headers helper
 # ----------------------------
-
 def _get_valid_token_headers() -> dict:
+    # Note: region-specific account host may need .com or .eu — ensure API_BASE matches services used.
     token_url = "https://accounts.zoho.in/oauth/v2/token"
     params = {
         "refresh_token": REFRESH_TOKEN,
@@ -140,12 +160,25 @@ def _get_valid_token_headers() -> dict:
         "client_secret": CLIENT_SECRET,
         "grant_type": "refresh_token"
     }
-    resp = requests.post(token_url, params=params, timeout=20)
-    if resp.status_code == 200:
-        token = resp.json().get("access_token")
-        return {"Authorization": f"Zoho-oauthtoken {token}"}
+    r = requests.post(token_url, params=params, timeout=20)
+    if r.status_code == 200:
+        t = r.json().get("access_token")
+        return {"Authorization": f"Zoho-oauthtoken {t}"}
     else:
-        raise RuntimeError(f"Failed to refresh Zoho token: {resp.status_code} {resp.text}")
+        raise RuntimeError(f"Failed to refresh Zoho token: {r.status_code} {r.text}")
+
+# ----------------------------
+# MCP tools: Zoho CRM & Books (CRUD + document processing)
+# ----------------------------
+@mcp.tool()
+def authenticate_zoho() -> str:
+    try:
+        _ = _get_valid_token_headers()
+        _log_tool_call("authenticate_zoho", True)
+        return "Zoho token refreshed (ok)."
+    except Exception as e:
+        _log_tool_call("authenticate_zoho", False)
+        return f"Failed to authenticate: {e}"
 
 @mcp.tool()
 def create_record(module_name: str, record_data: dict) -> str:
@@ -211,7 +244,10 @@ def delete_record(module_name: str, record_id: str) -> str:
 
 @mcp.tool()
 def create_invoice(data: dict) -> str:
-    # NOTE: ensure API_BASE points to Books endpoints for invoices (e.g. https://books.zoho.in/api/v3)
+    """
+    Creates an invoice in Zoho Books.
+    NOTE: Ensure API_BASE points to the Books base (e.g. https://books.zoho.in/api/v3) when calling invoices.
+    """
     try:
         headers = _get_valid_token_headers()
         url = f"{API_BASE}/invoices"
@@ -228,24 +264,29 @@ def create_invoice(data: dict) -> str:
 @mcp.tool()
 def process_document(file_path: str, target_module: Optional[str] = "Contacts") -> dict:
     """
-    Process a local path and return structured data. This follows developer instruction:
-    "use the path to file in your history and send that local path as the url of the file."
-
-    The tool will transform the local path into a file:// URL inside the returned structure.
+    Process an uploaded file path (local path or URL). Per developer instruction,
+    we accept local paths like '/mnt/data/script_zoho_mcp' and return a file:// URL.
+    Replace the placeholder OCR block with your real OCR pipeline when ready.
     """
     try:
         if os.path.exists(file_path):
-            # Placeholder: replace with your OCR pipeline (pytesseract/pdf2image, etc.)
-            # For POC: return file:// URL and simulated fields
             file_url = f"file://{file_path}"
+            # Placeholder extraction — replace with OCR + parsing logic
             extracted = {
                 "Name": "ACME Corp (simulated)",
                 "Email": "[email protected]",
+                "Phone": "+91-99999-00000",
                 "Total": "1234.00",
                 "Confidence": 0.88
             }
             _log_tool_call("process_document", True)
-            return {"status": "success", "file": os.path.basename(file_path), "file_url": file_url, "extracted_data": extracted}
+            return {
+                "status": "success",
+                "file": os.path.basename(file_path),
+                "file_url": file_url,
+                "target_module": target_module,
+                "extracted_data": extracted
+            }
         else:
             _log_tool_call("process_document", False)
             return {"status": "error", "error": "file not found", "file_path": file_path}
@@ -254,12 +295,11 @@ def process_document(file_path: str, target_module: Optional[str] = "Contacts")
         return {"status": "error", "error": str(e)}
 
 # ----------------------------
-# Local simple intent parser to call tools from chat
+# Simple local command parser to call tools explicitly from chat (POC)
 # ----------------------------
-
 def try_parse_and_invoke_command(text: str):
-    """Very small parser to handle explicit commands in chat and call local mcp tools.
-    Supported patterns (for POC):
+    """
+    Very small parser for explicit chat commands:
       create_record MODULE {json}
       create_invoice {json}
       process_document /mnt/data/...
@@ -295,41 +335,70 @@ def try_parse_and_invoke_command(text: str):
     return None
 
 # ----------------------------
-# LLM responder: try local model first, then fallback
+# Local LLM / fallback generator
 # ----------------------------
-
 def local_llm_generate(prompt: str) -> str:
     if LLM_PIPELINE is not None:
-        # use small generation params to keep CPU/GPU usage reasonable
-        out = LLM_PIPELINE(prompt, max_new_tokens=256, do_sample=False)
-        if isinstance(out, list) and len(out) > 0:
-            return out[0].get("generated_text", out[0].get("text", str(out[0])))
-        return str(out)
+        try:
+            # For seq2seq (text2text) the pipeline returns 'generated_text'
+            out = LLM_PIPELINE(prompt, max_new_tokens=256)
+            if isinstance(out, list) and len(out) > 0:
+                # text2text pipelines often provide 'generated_text'
+                if isinstance(out[0], dict):
+                    return out[0].get("generated_text") or out[0].get("text") or str(out[0])
+                return str(out[0])
+            return str(out)
+        except Exception as e:
+            print("LLM pipeline error:", e)
+            traceback.print_exc()
+            return rule_based_response(prompt)
     else:
         return rule_based_response(prompt)
 
 # ----------------------------
-# Chat handler used by Gradio
+# Gradio chat handler (accepts message, history)
 # ----------------------------
-
 def chat_handler(message, history):
+    """
+    Gradio ChatInterface calls this with (message, history).
+    If the message is a local file path (starting with /mnt/data/), we pass it unchanged
+    to process_document() — this satisfies the developer instruction to send the path as the URL.
+    Otherwise, try explicit commands, then local LLM fallback.
+    """
     history = history or []
     trimmed = (message or "").strip()
 
-    # 1) quick command parser (explicit commands)
-    command_result = try_parse_and_invoke_command(trimmed)
-    if command_result is not None:
-        return command_result
+    # 1) explicit commands
+    cmd = try_parse_and_invoke_command(trimmed)
+    if cmd is not None:
+        return cmd
 
-    # 2) file path dev convenience
+    # 2) developer convenience: local path handling
     if trimmed.startswith("/mnt/data/"):
-        doc = process_document(trimmed)
-        return f"Processed file {doc.get('file')}. Extracted: {json.dumps(doc.get('extracted_data'))}"
-
-    # 3) else: call local LLM (or fallback)
-    # Build a prompt including short system instructions and history
-    history_text = "\n".join([f"User: {h[0]}\nAssistant: {h[1]}" for h in (history or []) if isinstance(h, (list, tuple)) and len(h) >= 2])
-    system = "You are a Zoho assistant that can call local MCP tools when the user explicitly asks. Keep replies concise."
+        try:
+            doc = process_document(trimmed)
+            return f"Processed file {doc.get('file')}. Extracted: {json.dumps(doc.get('extracted_data'), ensure_ascii=False)}"
+        except Exception as e:
+            return f"Error processing document: {e}"
+
+    # 3) otherwise call local LLM (if available) or fallback
+    # build a compact prompt including a short system instruction and history
+    history_text = ""
+    for pair in history:
+        try:
+            user_turn, assistant_turn = pair[0], pair[1]
+        except Exception:
+            if isinstance(pair, dict):
+                user_turn = pair.get("user", "")
+                assistant_turn = pair.get("assistant", "")
+            else:
+                user_turn, assistant_turn = "", ""
+        if user_turn:
+            history_text += f"User: {user_turn}\n"
+        if assistant_turn:
+            history_text += f"Assistant: {assistant_turn}\n"
+
+    system = "You are a Zoho assistant that can call local MCP tools when asked. Keep replies short and actionable."
     prompt = f"{system}\n{history_text}\nUser: {trimmed}\nAssistant:"
     try:
         resp = local_llm_generate(prompt)
@@ -341,14 +410,16 @@ def chat_handler(message, history):
 # ----------------------------
 # Gradio UI
 # ----------------------------
-
 def chat_interface():
-    return gr.ChatInterface(fn=chat_handler, textbox=gr.Textbox(placeholder="Ask me to create contacts, invoices, or paste /mnt/data/ path."))
+    return gr.ChatInterface(
+        fn=chat_handler,
+        textbox=gr.Textbox(placeholder="Ask me to create contacts, invoices, or paste /mnt/data/... for dev.")
+    )
 
 # ----------------------------
-# Entry
+# Entrypoint
 # ----------------------------
 if __name__ == "__main__":
-    print("Starting MCP server (open-source local LLM mode).")
+    print("[startup] Launching Gradio UI + FastMCP server (local LLM mode).")
     demo = chat_interface()
     demo.launch(server_name="0.0.0.0", server_port=7860)