Initial commit - Secure Reasoning MCP Server

README.md

@@ -0,0 +1,65 @@
+---
+title: Secure Reasoning MCP Server
+emoji: 🛡️
+colorFrom: purple
+colorTo: green
+sdk: gradio
+sdk_version: 5.0.0
+app_file: app.py
+pinned: false
+tags:
+  - mcp-in-action-track-enterprise
+  - agent
+  - security
+  - langgraph
+  - merkle-tree
+---
+
+# 🛡️ Secure Reasoning MCP Server
+> **"Don't Trust, Verify."** — AI Ajanları için Şeffaf, Denetlenebilir ve Değiştirilemez Muhakeme (Reasoning) Katmanı.
+
+[![Hugging Face Spaces](https://img.shields.io/badge/🤗%20Hugging%20Face-Spaces-blue)](https://huggingface.co/spaces)
+[![Built with Gradio](https://img.shields.io/badge/Built%20with-Gradio%205-orange)](https://gradio.app)
+[![Powered by LangGraph](https://img.shields.io/badge/Powered%20by-LangGraph-green)](https://langchain-ai.github.io/langgraph/)
+
+## 🏆 Hackathon Track
+Bu proje **MCP 1st Birthday Hackathon** kapsamında geliştirilmiştir.
+* **Track:** `Track 2: MCP in Action`
+* **Category Tag:** `mcp-in-action-track-enterprise`
+*(Not: Bu proje kurumsal yapay zeka güvenliği ve denetlenebilirlik (audit) sorunlarına çözüm getirdiği için Enterprise kategorisindedir.)*
+
+## 💡 Problem: Kara Kutu Sorunu
+Otonom AI ajanları (Agents) giderek daha karmaşık görevleri yerine getiriyor. Ancak kritik bir sorun var: **Bir ajanın neden o kararı verdiğini veya işlem sırasında manipüle edilip edilmediğini nasıl kanıtlayabilirsiniz?**
+Mevcut sistemlerde loglar silinebilir, değiştirilebilir ve ajanın düşünce süreci (reasoning chain) bir kara kutudur.
+
+## 🚀 Çözüm: Kriptografik "Chain-of-Checks"
+Biz, sadece "Chain-of-Thought" (Düşünce Zinciri) değil, **"Chain-of-Checks" (Denetim Zinciri)** sunuyoruz.
+
+Sistemimiz iki ana katmandan oluşur:
+1.  **The Brain (LangGraph):** Planlayan, güvenlik kontrolü yapan ve uygulayan zeka.
+2.  **The Ledger (Crypto Engine):** Her adımı hash'leyen, Merkle Ağacına ekleyen ve WORM (Write-Once-Read-Many) depolamada saklayan kasa.
+
+## 🏗️ Mimari (Architecture)
+
+Sistemimiz **Gradio 5** arayüzü arkasında çalışan, olay tabanlı (event-driven) bir LangGraph mimarisi kullanır.
+
+```mermaid
+graph TD
+    User[Kullanıcı Görevi] -->|Gradio UI| Agent
+    
+    subgraph "🛡️ Secure Agent (The Brain)"
+        Agent --> Plan[📝 Planner Node]
+        Plan --> Safety{🛡️ Safety Check}
+        Safety -->|Riskli| Refine[Refiner Node]
+        Safety -->|Güvenli| Exec[⚡ Executor Node]
+        Exec --> Justify[💭 Justification]
+    end
+    
+    subgraph "🔒 Immutable Ledger (The Vault)"
+        Exec -.->|Log Data| Hash[#️⃣ SHA-256 Hash]
+        Hash --> Merkle[🌳 Merkle Tree Update]
+        Merkle --> WORM[💾 WORM Storage]
+        WORM -.->|Cryptographic Proof| UI[Gradio Dashboard]
+    end
+    
+    Justify -->|Stream| UI

app.py

@@ -0,0 +1,148 @@
+import gradio as gr
+import os
+from dotenv import load_dotenv
+import asyncio
+import uuid
+
+# Your modules
+from graph import create_reasoning_graph
+from state import create_initial_state
+
+# Load environment variables
+load_dotenv()
+
+# Initialize the graph
+graph = create_reasoning_graph()
+
+
+async def run_agent_stream(user_input):
+    if not user_input or not user_input.strip():
+        yield "Please enter a task.", "SYSTEM READY"
+        return
+    
+    reasoning_log = "Starting Secure Reasoning Pipeline...\n\n"
+    crypto_log = "SECURE LEDGER INITIALIZED...\nWaiting for execution...\n"
+    
+    yield reasoning_log, crypto_log
+
+    task_id = f"task_{uuid.uuid4().hex[:8]}"
+    config = {"configurable": {"thread_id": task_id}}
+    
+    initial_state = create_initial_state(
+        task=user_input.strip(),
+        task_id=task_id,
+        user_id="gradio_user"
+    )
+    
+    try:
+        async for event in graph.astream(initial_state, config, stream_mode="values"):
+            
+            if "messages" in event and event["messages"]:
+                last_msg = event["messages"][-1]
+                if hasattr(last_msg, "content") and hasattr(last_msg, "type"):
+                    if last_msg.type == "ai":
+                        content = last_msg.content[:500] if len(last_msg.content) > 500 else last_msg.content
+                        reasoning_log += f"\nAgent: {content}\n"
+            
+            if "plan" in event and event["plan"]:
+                plan = event["plan"]
+                if hasattr(plan, "steps") and plan.steps:
+                    plan_text = "\nExecution Plan:\n"
+                    for step in plan.steps:
+                        plan_text += f"  - Step {step.step_number}: {step.action}\n"
+                    if plan_text not in reasoning_log:
+                        reasoning_log += plan_text
+            
+            if "status" in event:
+                status = event["status"]
+                if status == "executing":
+                    step_idx = event.get("current_step_index", 0)
+                    reasoning_log += f"\nExecuting Step {step_idx + 1}...\n"
+                elif status == "completed":
+                    reasoning_log += "\nTask Completed Successfully!\n"
+                elif status == "blocked":
+                    reasoning_log += "\nTask Blocked by Safety Guardrails\n"
+                elif status == "failed":
+                    error = event.get("error", "Unknown error")
+                    reasoning_log += f"\nTask Failed: {error}\n"
+
+            if "logs" in event and event["logs"]:
+                latest_log = event["logs"][-1]
+                
+                timestamp = getattr(latest_log, "timestamp", "N/A")
+                action_hash = getattr(latest_log, "action_hash", "N/A")
+                merkle_root = getattr(latest_log, "merkle_root", "N/A")
+                worm_path = getattr(latest_log, "worm_path", "memory")
+                step_number = getattr(latest_log, "step_number", "?")
+                
+                if hasattr(timestamp, "isoformat"):
+                    timestamp = timestamp.isoformat()
+                
+                log_entry = f"\n--------------------------------------------------\nSTEP: {step_number}\nTIME: {timestamp}\nHASH: {str(action_hash)[:20]}...\nROOT: {str(merkle_root)[:20]}...\nWORM: {worm_path}\nPROOF VERIFIED\n--------------------------------------------------\n"
+                
+                if str(action_hash)[:20] not in crypto_log:
+                    crypto_log += log_entry
+
+            yield reasoning_log, crypto_log
+            
+    except Exception as e:
+        reasoning_log += f"\n\nError: {str(e)}\n"
+        yield reasoning_log, crypto_log
+
+
+custom_css = '''
+#reasoning-box {
+    height: 500px; 
+    overflow-y: scroll; 
+    background-color: #f9f9f9; 
+    border: 1px solid #ddd;
+    padding: 15px;
+}
+#crypto-box {
+    height: 500px; 
+    overflow-y: scroll; 
+    background-color: #1e1e1e; 
+    color: #00ff00; 
+    font-family: monospace;
+    border: 1px solid #333;
+    padding: 15px;
+}
+'''
+
+with gr.Blocks(theme=gr.themes.Soft(), css=custom_css, title="Secure Reasoning MCP") as demo:
+    
+    gr.Markdown("# Secure Reasoning MCP Server\n**Verifier:** Gradio 5 + LangGraph + Merkle Trees")
+    
+    with gr.Row():
+        with gr.Column(scale=1):
+            user_input = gr.Textbox(
+                label="Task Description",
+                placeholder="E.g.: Write a short report about renewable energy...",
+                lines=2
+            )
+            submit_btn = gr.Button("Start Task", variant="primary")
+            
+    with gr.Row():
+        with gr.Column(scale=1):
+            gr.Markdown("### Agent Reasoning Flow")
+            reasoning_output = gr.Markdown(elem_id="reasoning-box", value="Waiting for task...")
+            
+        with gr.Column(scale=1):
+            gr.Markdown("### Immutable Crypto Ledger")
+            crypto_output = gr.Textbox(
+                elem_id="crypto-box", 
+                value="SYSTEM READY", 
+                lines=20, 
+                max_lines=20,
+                show_label=False,
+                interactive=False
+            )
+
+    submit_btn.click(
+        fn=run_agent_stream,
+        inputs=[user_input],
+        outputs=[reasoning_output, crypto_output]
+    )
+
+if __name__ == "__main__":
+    demo.launch()

crypto_engine.py

@@ -0,0 +1,333 @@
+import hashlib
+import json
+import time
+import os
+
+
+def hash_tool(data):
+    """
+    Hash tool: Convert input to deterministic string and return SHA-256 hex digest.
+    """
+    if isinstance(data, dict):
+        data_str = json.dumps(data, sort_keys=True)
+    else:
+        data_str = str(data)
+    
+    return hashlib.sha256(data_str.encode()).hexdigest()
+
+
+class MerkleTree:
+    """
+    Merkle Tree: Maintains a list of leaves and recalculates root on each update.
+    """
+    def __init__(self):
+        self.leaves = []
+        self.root = None
+    
+    def _calculate_root(self, leaves):
+        """
+        Calculate Merkle root from a list of leaves.
+        If empty, return None. If single leaf, return it. Otherwise, build tree bottom-up.
+        """
+        if not leaves:
+            return None
+        
+        if len(leaves) == 1:
+            return leaves[0]
+        
+        current_level = leaves[:]
+        
+        while len(current_level) > 1:
+            next_level = []
+            for i in range(0, len(current_level), 2):
+                left = current_level[i]
+                right = current_level[i + 1] if i + 1 < len(current_level) else left
+                combined = hashlib.sha256((left + right).encode()).hexdigest()
+                next_level.append(combined)
+            current_level = next_level
+        
+        return current_level[0]
+    
+    def update(self, new_hash):
+        """
+        Add a new leaf and recalculate the Merkle root.
+        Returns the new root.
+        """
+        self.leaves.append(new_hash)
+        self.root = self._calculate_root(self.leaves)
+        return self.root
+    
+    def get_proof(self, index):
+        """
+        Generate Merkle proof for a leaf at given index.
+        Returns a list of (sibling_hash, position) tuples where position is 'left' or 'right'.
+        """
+        if index < 0 or index >= len(self.leaves):
+            return None
+        
+        proof = []
+        current_index = index
+        current_level = self.leaves[:]
+        
+        while len(current_level) > 1:
+            # Determine if current_index is odd (right) or even (left)
+            is_right = current_index % 2 == 1
+            sibling_index = current_index - 1 if is_right else current_index + 1
+            
+            # Get sibling hash if it exists
+            if sibling_index < len(current_level):
+                sibling_hash = current_level[sibling_index]
+                position = "left" if is_right else "right"
+                proof.append({"hash": sibling_hash, "position": position})
+            
+            # Move to next level
+            current_index = current_index // 2
+            next_level = []
+            for i in range(0, len(current_level), 2):
+                left = current_level[i]
+                right = current_level[i + 1] if i + 1 < len(current_level) else left
+                combined = hashlib.sha256((left + right).encode()).hexdigest()
+                next_level.append(combined)
+            current_level = next_level
+        
+        return proof
+
+
+def worm_write_tool(step_data, hash_value, merkle_root, filename="worm_log.jsonl"):
+    """
+    WORM (Write Once, Read Many) storage: Append a JSON record to JSONL file.
+    Returns the record written.
+    """
+    # Determine the next ID by counting existing lines
+    next_id = 0
+    if os.path.exists(filename):
+        with open(filename, "r") as f:
+            next_id = sum(1 for _ in f)
+    
+    record = {
+        "id": next_id,
+        "timestamp": time.time(),
+        "step": step_data,
+        "hash": hash_value,
+        "root": merkle_root
+    }
+    
+    with open(filename, "a") as f:
+        f.write(json.dumps(record) + "\n")
+    
+    return record
+
+
+def proof_generate_tool(record_id, filename="worm_log.jsonl"):
+    """
+    Generate a Merkle proof for a specific record in the WORM log.
+    Rehydrates the Merkle Tree from the log to ensure proof is against current state.
+    Returns a JSON proof containing hash, merkle_proof, root, and timestamp.
+    """
+    if not os.path.exists(filename):
+        print(f"[PROOF] Error: {filename} does not exist.")
+        return None
+    
+    # Read all records from WORM log
+    records = []
+    hashes = []
+    target_record = None
+    
+    with open(filename, "r") as f:
+        for line in f:
+            record = json.loads(line.strip())
+            records.append(record)
+            hashes.append(record["hash"])
+            if record["id"] == record_id:
+                target_record = record
+    
+    if target_record is None:
+        print(f"[PROOF] Error: Record with ID {record_id} not found.")
+        return None
+    
+    # Rehydrate Merkle Tree from hashes
+    tree = MerkleTree()
+    for h in hashes:
+        tree.update(h)
+    
+    # Get proof for the target record index
+    proof = tree.get_proof(record_id)
+    
+    proof_result = {
+        "record_id": record_id,
+        "hash": target_record["hash"],
+        "merkle_proof": proof,
+        "merkle_root": tree.root,
+        "timestamp": target_record["timestamp"],
+        "step_details": target_record["step"]
+    }
+    
+    return proof_result
+
+
+def verify_proof_tool(target_hash, merkle_proof, merkle_root):
+    """
+    Verify if a target_hash belongs to the merkle_root using the merkle_proof.
+    
+    Logic:
+    - Start with current_hash = target_hash
+    - Loop through proof items (sibling hashes with positions)
+    - Reconstruct the path up to the root
+    - Compare final calculated root with provided merkle_root
+    
+    Returns True if valid, False otherwise.
+    """
+    if merkle_proof is None:
+        return False
+    
+    current_hash = target_hash
+    
+    # Traverse the proof path
+    for proof_item in merkle_proof:
+        sibling_hash = proof_item["hash"]
+        position = proof_item["position"]
+        
+        # Combine hashes based on position
+        if position == "left":
+            # Sibling is on the left, so: hash(sibling + current)
+            combined_str = sibling_hash + current_hash
+        elif position == "right":
+            # Sibling is on the right, so: hash(current + sibling)
+            combined_str = current_hash + sibling_hash
+        else:
+            return False
+        
+        # Calculate the next level hash
+        current_hash = hashlib.sha256(combined_str.encode()).hexdigest()
+    
+    # Final check: does calculated root match provided root?
+    return current_hash == merkle_root
+
+
+def secure_agent_action(action_type, details, merkle_tree):
+    """
+    Gatekeeper Logic: Cite-Before-Act mechanism.
+    - READ: Auto-approve
+    - WRITE/MUTATE: Require human approval via CLI
+    All actions (approved or denied) are logged to WORM storage.
+    """
+    action_type = action_type.upper()
+    
+    if action_type == "READ":
+        # Auto-approve READ actions
+        print(f"\n[GATEKEEPER] READ action detected: {details}")
+        print("[GATEKEEPER] Auto-approving READ action.")
+        
+        step_data = {
+            "action_type": action_type,
+            "details": details,
+            "status": "APPROVED"
+        }
+        
+        step_hash = hash_tool(step_data)
+        merkle_root = merkle_tree.update(step_hash)
+        worm_write_tool(step_data, step_hash, merkle_root)
+        
+        print(f"[GATEKEEPER] Merkle Root: {merkle_root}")
+        print(f"[GATEKEEPER] Action logged.\n")
+        return True
+    
+    elif action_type in ["WRITE", "MUTATE", "DELETE"]:
+        # Require approval for mutation actions
+        print(f"\n[GATEKEEPER] ⚠️  WRITE/MUTATE action detected: {details}")
+        print("[GATEKEEPER] This action requires human approval.")
+        
+        approval = input("[GATEKEEPER] Approve this action? (y/n): ").strip().lower()
+        
+        if approval == "y":
+            print("[GATEKEEPER] ✓ Action APPROVED by user.")
+            status = "APPROVED"
+            result = True
+        else:
+            print("[GATEKEEPER] ✗ Action DENIED by user.")
+            status = "DENIED"
+            result = False
+        
+        # Log the action (approved or denied) to maintain audit trail
+        step_data = {
+            "action_type": action_type,
+            "details": details,
+            "status": status
+        }
+        
+        step_hash = hash_tool(step_data)
+        merkle_root = merkle_tree.update(step_hash)
+        worm_write_tool(step_data, step_hash, merkle_root)
+        
+        print(f"[GATEKEEPER] Merkle Root: {merkle_root}")
+        print(f"[GATEKEEPER] Audit logged.\n")
+        return result
+    
+    else:
+        print(f"\n[GATEKEEPER] Unknown action type: {action_type}\n")
+        return False
+
+
+if __name__ == "__main__":
+    print("=" * 70)
+    print("SECURE REASONING MCP SERVER - TEST SCENARIO")
+    print("=" * 70)
+    
+    # Initialize Merkle Tree
+    mt = MerkleTree()
+    
+    # Test 1: READ action (auto-approved)
+    print("\n[TEST 1] Simulating READ action...")
+    secure_agent_action("READ", "Query user database for profile info", mt)
+    
+    # Test 2: WRITE action (user approval - simulate "y")
+    print("[TEST 2] Simulating WRITE action (approve with 'y')...")
+    secure_agent_action("WRITE", "Update user profile with new email address", mt)
+    
+    # Test 3: WRITE action (user denial - simulate "n")
+    print("[TEST 3] Simulating WRITE action (deny with 'n')...")
+    secure_agent_action("WRITE", "Delete user account permanently", mt)
+    
+    print("=" * 70)
+    print("TEST SCENARIO COMPLETE")
+    print("=" * 70)
+    print("\nWORM Log saved to: worm_log.jsonl")
+    print("Review the file to verify all actions are logged with hashes and Merkle roots.\n")
+    
+    # Test 4: Generate proof for record_id=1
+    print("=" * 70)
+    print("PROOF GENERATION TEST")
+    print("=" * 70)
+    print("\n[TEST 4] Generating Merkle proof for record_id=1...")
+    proof = proof_generate_tool(1)
+    if proof:
+        print("\n[PROOF] Generated Merkle Proof:")
+        print(json.dumps(proof, indent=2))
+    else:
+        proof = None
+    print()
+    
+    # Test 5: Verify the proof (positive case)
+    print("=" * 70)
+    print("PROOF VERIFICATION TEST")
+    print("=" * 70)
+    if proof:
+        print("\n[TEST 5a] Verifying proof with correct hash and root...")
+        is_valid = verify_proof_tool(proof["hash"], proof["merkle_proof"], proof["merkle_root"])
+        print(f"[VERIFY] Verification Result (POSITIVE): {is_valid}")
+        
+        # Test 5b: Verify with tampered hash (negative case)
+        print("\n[TEST 5b] Verifying proof with tampered hash (should fail)...")
+        tampered_hash = proof["hash"][:-2] + "XX"  # Change last 2 characters
+        is_valid_tampered = verify_proof_tool(tampered_hash, proof["merkle_proof"], proof["merkle_root"])
+        print(f"[VERIFY] Verification Result (NEGATIVE - tampered hash): {is_valid_tampered}")
+        
+        # Test 5c: Verify with tampered root (negative case)
+        print("\n[TEST 5c] Verifying proof with tampered root (should fail)...")
+        tampered_root = proof["merkle_root"][:-2] + "XX"  # Change last 2 characters
+        is_valid_tampered_root = verify_proof_tool(proof["hash"], proof["merkle_proof"], tampered_root)
+        print(f"[VERIFY] Verification Result (NEGATIVE - tampered root): {is_valid_tampered_root}")
+    
+    print("\n" + "=" * 70)
+    print("ALL TESTS COMPLETE - SECURE REASONING MCP SERVER OPERATIONAL")
+    print("=" * 70 + "\n")

graph.py

@@ -0,0 +1,702 @@
+"""
+LangGraph State Machine for Secure Reasoning MCP Server
+Implements the Chain-of-Checks workflow with cryptographic logging.
+"""
+import json
+from typing import Literal
+from datetime import datetime
+
+from langgraph.graph import StateGraph, END
+from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
+from langchain_anthropic import ChatAnthropic
+
+from state import AgentState
+from schemas import (
+    ExecutionPlan, StepPlan, SafetyCheckResult, ExecutionResult,
+    Justification, CryptoLogEntry, HashRequest, MerkleUpdateRequest,
+    WORMWriteRequest
+)
+from prompts import (
+    format_planner_prompt, format_safety_prompt, format_executor_prompt,
+    format_justification_prompt, format_synthesis_prompt
+)
+from mock_tools import MockCryptoTools
+
+
+# ============================================================================
+# GLOBAL CONFIGURATION
+# ============================================================================
+
+# Initialize LLM (Claude 3.5 Sonnet)
+llm = ChatAnthropic(
+    model="claude-3-5-sonnet-20241022",
+    temperature=0.1,  # Low temperature for deterministic reasoning
+    max_tokens=4096
+)
+
+# Initialize crypto tools (replace with real tools when ready)
+crypto_tools = MockCryptoTools()
+
+
+# ============================================================================
+# NODE 1: PLANNER
+# ============================================================================
+
+def planner_node(state: AgentState) -> AgentState:
+    """
+    Generate a step-by-step execution plan for the task.
+    
+    Args:
+        state: Current agent state with the task
+        
+    Returns:
+        Updated state with the execution plan
+    """
+    print(f"\n{'='*60}")
+    print(f"🧠 PLANNER NODE - Generating execution plan")
+    print(f"{'='*60}")
+    
+    # Format the prompt
+    prompts = format_planner_prompt(state["task"])
+    
+    # Create messages
+    messages = [
+        SystemMessage(content=prompts["system"]),
+        HumanMessage(content=prompts["user"])
+    ]
+    
+    # Call LLM
+    response = llm.invoke(messages)
+    
+    # Parse JSON response
+    try:
+        plan_data = json.loads(response.content)
+        
+        # Convert to ExecutionPlan model
+        steps = [StepPlan(**step) for step in plan_data["steps"]]
+        plan = ExecutionPlan(
+            steps=steps,
+            total_steps=plan_data["total_steps"]
+        )
+        
+        print(f"✅ Generated plan with {plan.total_steps} steps:")
+        for step in steps:
+            print(f"   Step {step.step_number}: {step.action}")
+        
+        # Update state
+        state["plan"] = plan
+        state["current_step_index"] = 0
+        state["status"] = "executing"
+        state["messages"].extend([
+            HumanMessage(content=prompts["user"]),
+            AIMessage(content=response.content)
+        ])
+        
+        return state
+        
+    except json.JSONDecodeError as e:
+        print(f"❌ Failed to parse planner response: {e}")
+        state["error"] = f"Planner failed to generate valid JSON: {str(e)}"
+        state["status"] = "failed"
+        return state
+
+
+# ============================================================================
+# NODE 2: SAFETY CHECKER
+# ============================================================================
+
+def safety_node(state: AgentState) -> AgentState:
+    """
+    Validate that the current step is safe to execute.
+    
+    Args:
+        state: Current agent state with the plan
+        
+    Returns:
+        Updated state with safety validation result
+    """
+    print(f"\n{'='*60}")
+    print(f"🛡️  SAFETY NODE - Validating step {state['current_step_index'] + 1}")
+    print(f"{'='*60}")
+    
+    # Get current step
+    current_step = state["plan"].steps[state["current_step_index"]]
+    
+    # Format previous steps for context
+    previous_steps = "None"
+    if state["current_step_index"] > 0:
+        prev_steps_list = [
+            f"Step {i+1}: {state['plan'].steps[i].action}"
+            for i in range(state["current_step_index"])
+        ]
+        previous_steps = "\n".join(prev_steps_list)
+    
+    # Format the prompt
+    prompts = format_safety_prompt(
+        step_description=current_step.action,
+        task=state["task"],
+        step_number=state["current_step_index"] + 1,
+        total_steps=state["plan"].total_steps,
+        previous_steps=previous_steps,
+        additional_context="This is a secure reasoning system with cryptographic logging."
+    )
+    
+    # Create messages
+    messages = [
+        SystemMessage(content=prompts["system"]),
+        HumanMessage(content=prompts["user"])
+    ]
+    
+    # Call LLM
+    response = llm.invoke(messages)
+    
+    # Parse JSON response
+    try:
+        safety_data = json.loads(response.content)
+        safety_result = SafetyCheckResult(**safety_data)
+        
+        print(f"🔍 Safety Check Result:")
+        print(f"   Is Safe: {safety_result.is_safe}")
+        print(f"   Risk Level: {safety_result.risk_level}")
+        print(f"   Reasoning: {safety_result.reasoning[:100]}...")
+        
+        # Update state
+        state["safety_status"] = safety_result
+        state["messages"].extend([
+            HumanMessage(content=prompts["user"]),
+            AIMessage(content=response.content)
+        ])
+        
+        # Mark if blocked
+        if not safety_result.is_safe:
+            state["safety_blocked"] = True
+            print(f"🚫 Step BLOCKED due to safety concerns")
+        else:
+            print(f"✅ Step approved for execution")
+        
+        return state
+        
+    except json.JSONDecodeError as e:
+        print(f"❌ Failed to parse safety response: {e}")
+        # Default to blocking if parsing fails (fail-safe)
+        state["safety_status"] = SafetyCheckResult(
+            is_safe=False,
+            risk_level="critical",
+            reasoning=f"Safety check failed due to parsing error: {str(e)}",
+            blocked_reasons=["parsing_error"]
+        )
+        state["safety_blocked"] = True
+        return state
+
+
+# ============================================================================
+# NODE 3: EXECUTOR
+# ============================================================================
+
+def executor_node(state: AgentState) -> AgentState:
+    """
+    Execute the current step (call tools if needed).
+    
+    Args:
+        state: Current agent state with approved step
+        
+    Returns:
+        Updated state with execution result
+    """
+    print(f"\n{'='*60}")
+    print(f"⚡ EXECUTOR NODE - Executing step {state['current_step_index'] + 1}")
+    print(f"{'='*60}")
+    
+    # Get current step
+    current_step = state["plan"].steps[state["current_step_index"]]
+    
+    # Format previous results for context
+    previous_results = "None"
+    if state["justifications"]:
+        prev_results_list = [
+            f"Step {j.step_number}: {j.reasoning[:100]}..."
+            for j in state["justifications"][-3:]  # Last 3 steps
+        ]
+        previous_results = "\n".join(prev_results_list)
+    
+    # Format the prompt
+    prompts = format_executor_prompt(
+        step_description=current_step.action,
+        task=state["task"],
+        expected_outcome=current_step.expected_outcome,
+        requires_tools=current_step.requires_tools,
+        previous_results=previous_results
+    )
+    
+    # Create messages
+    messages = [
+        SystemMessage(content=prompts["system"]),
+        HumanMessage(content=prompts["user"])
+    ]
+    
+    # Call LLM
+    response = llm.invoke(messages)
+    
+    # Parse JSON response
+    try:
+        executor_data = json.loads(response.content)
+        tool_needed = executor_data.get("tool_needed", "internal_reasoning")
+        tool_params = executor_data.get("tool_params")
+        direct_result = executor_data.get("direct_result")
+        
+        print(f"🔧 Tool Selection: {tool_needed}")
+        
+        # Execute based on tool selection
+        if tool_needed == "internal_reasoning":
+            result = ExecutionResult(
+                success=True,
+                output=direct_result or "Analysis completed through reasoning",
+                tool_calls=["internal_reasoning"]
+            )
+        else:
+            # Simulate tool execution (in real system, dispatch to actual tools)
+            result = ExecutionResult(
+                success=True,
+                output=f"Simulated result from {tool_needed} with params: {tool_params}",
+                tool_calls=[tool_needed]
+            )
+        
+        print(f"✅ Execution successful")
+        print(f"   Output: {str(result.output)[:100]}...")
+        
+        # Update state
+        state["execution_result"] = result
+        state["messages"].extend([
+            HumanMessage(content=prompts["user"]),
+            AIMessage(content=response.content)
+        ])
+        
+        return state
+        
+    except json.JSONDecodeError as e:
+        print(f"❌ Execution failed: {e}")
+        state["execution_result"] = ExecutionResult(
+            success=False,
+            output=None,
+            error=f"Failed to parse executor response: {str(e)}",
+            tool_calls=[]
+        )
+        return state
+    except Exception as e:
+        print(f"❌ Execution error: {e}")
+        state["execution_result"] = ExecutionResult(
+            success=False,
+            output=None,
+            error=str(e),
+            tool_calls=[]
+        )
+        return state
+
+
+# ============================================================================
+# NODE 4: LOGGER (Cryptographic Logging)
+# ============================================================================
+
+def logger_node(state: AgentState) -> AgentState:
+    """
+    Hash the execution result and log it to Merkle Tree + WORM storage.
+    
+    Args:
+        state: Current agent state with execution result
+        
+    Returns:
+        Updated state with cryptographic log entry
+    """
+    print(f"\n{'='*60}")
+    print(f"📝 LOGGER NODE - Creating cryptographic proof")
+    print(f"{'='*60}")
+    
+    current_step = state["plan"].steps[state["current_step_index"]]
+    execution_result = state["execution_result"]
+    
+    try:
+        # 1. Prepare the data to log
+        log_data = {
+            "task_id": state["task_id"],
+            "step_number": state["current_step_index"] + 1,
+            "action": current_step.action,
+            "result": execution_result.output if execution_result.success else execution_result.error,
+            "timestamp": datetime.utcnow().isoformat(),
+            "safety_approved": state["safety_status"].is_safe if state["safety_status"] else False
+        }
+        
+        # 2. Hash the action data
+        hash_request = HashRequest(
+            data=json.dumps(log_data, sort_keys=True),
+            algorithm="sha256"
+        )
+        hash_response = crypto_tools.hash_tool(hash_request)
+        action_hash = hash_response.hash
+        print(f"🔐 Action Hash: {action_hash[:16]}...")
+        
+        # 3. Update Merkle Tree
+        merkle_request = MerkleUpdateRequest(
+            leaf_hash=action_hash,
+            metadata={"step": state["current_step_index"] + 1}
+        )
+        merkle_response = crypto_tools.merkle_update_tool(merkle_request)
+        merkle_root = merkle_response.merkle_root
+        print(f"🌳 Merkle Root: {merkle_root[:16]}...")
+        
+        # 4. Write to WORM storage
+        entry_id = f"{state['task_id']}_step_{state['current_step_index'] + 1}"
+        worm_request = WORMWriteRequest(
+            entry_id=entry_id,
+            data=log_data,
+            merkle_root=merkle_root
+        )
+        worm_response = crypto_tools.worm_write_tool(worm_request)
+        print(f"💾 WORM Path: {worm_response.storage_path}")
+        
+        # 5. Create log entry
+        log_entry = CryptoLogEntry(
+            step_number=state["current_step_index"] + 1,
+            action_hash=action_hash,
+            merkle_root=merkle_root,
+            worm_path=worm_response.storage_path
+        )
+        
+        # Update state
+        state["logs"].append(log_entry)
+        print(f"✅ Cryptographic logging complete")
+        
+        return state
+        
+    except Exception as e:
+        print(f"❌ Logging failed: {e}")
+        state["error"] = f"Cryptographic logging failed: {str(e)}"
+        return state
+
+
+# ============================================================================
+# NODE 5: JUSTIFICATION
+# ============================================================================
+
+def justification_node(state: AgentState) -> AgentState:
+    """
+    Generate an explanation for why the action was taken.
+    
+    Args:
+        state: Current agent state with execution result
+        
+    Returns:
+        Updated state with justification
+    """
+    print(f"\n{'='*60}")
+    print(f"💭 JUSTIFICATION NODE - Explaining the action")
+    print(f"{'='*60}")
+    
+    current_step = state["plan"].steps[state["current_step_index"]]
+    execution_result = state["execution_result"]
+    
+    # Determine tool used
+    tool_used = ", ".join(execution_result.tool_calls) if execution_result.tool_calls else "none"
+    
+    # Format the prompt
+    prompts = format_justification_prompt(
+        step_description=current_step.action,
+        tool_used=tool_used,
+        execution_result=str(execution_result.output)[:500] if execution_result.success else execution_result.error,
+        task=state["task"],
+        step_number=state["current_step_index"] + 1,
+        total_steps=state["plan"].total_steps,
+        expected_outcome=current_step.expected_outcome
+    )
+    
+    # Create messages
+    messages = [
+        SystemMessage(content=prompts["system"]),
+        HumanMessage(content=prompts["user"])
+    ]
+    
+    # Call LLM
+    response = llm.invoke(messages)
+    
+    # Parse JSON response
+    try:
+        justification_data = json.loads(response.content)
+        justification = Justification(**justification_data)
+        
+        print(f"📋 Justification generated:")
+        print(f"   {justification.reasoning[:150]}...")
+        
+        # Update state
+        state["justifications"].append(justification)
+        state["messages"].extend([
+            HumanMessage(content=prompts["user"]),
+            AIMessage(content=response.content)
+        ])
+        
+        return state
+        
+    except json.JSONDecodeError as e:
+        print(f"⚠️  Failed to parse justification, using fallback: {e}")
+        # Create fallback justification
+        fallback = Justification(
+            step_number=state["current_step_index"] + 1,
+            reasoning=f"Executed {current_step.action} as planned. Result: {execution_result.success}",
+            evidence=None,
+            alternatives_considered=None
+        )
+        state["justifications"].append(fallback)
+        return state
+
+
+# ============================================================================
+# NODE 6: STEP ITERATOR
+# ============================================================================
+
+def step_iterator_node(state: AgentState) -> AgentState:
+    """
+    Move to the next step or complete the task.
+    
+    Args:
+        state: Current agent state
+        
+    Returns:
+        Updated state with incremented step index
+    """
+    print(f"\n{'='*60}")
+    print(f"➡️  STEP ITERATOR - Moving to next step")
+    print(f"{'='*60}")
+    
+    # Increment step index
+    state["current_step_index"] += 1
+    
+    # Check if we're done
+    if state["current_step_index"] >= state["plan"].total_steps:
+        print(f"🎉 All steps completed!")
+        state["status"] = "completed"
+    else:
+        print(f"📍 Moving to step {state['current_step_index'] + 1}/{state['plan'].total_steps}")
+    
+    return state
+
+
+# ============================================================================
+# NODE 7: REFINER (for unsafe steps)
+# ============================================================================
+
+def refiner_node(state: AgentState) -> AgentState:
+    """
+    Handle unsafe steps by modifying or skipping them.
+    
+    Args:
+        state: Current agent state with blocked step
+        
+    Returns:
+        Updated state with refinement decision
+    """
+    print(f"\n{'='*60}")
+    print(f"🔧 REFINER NODE - Handling unsafe step")
+    print(f"{'='*60}")
+    
+    current_step = state["plan"].steps[state["current_step_index"]]
+    safety_status = state["safety_status"]
+    
+    # Log the blocked action
+    print(f"🚫 Step blocked: {current_step.action}")
+    print(f"   Reason: {safety_status.reasoning}")
+    
+    # Create a null execution result
+    state["execution_result"] = ExecutionResult(
+        success=False,
+        output=None,
+        error=f"Step blocked by safety guardrails: {safety_status.reasoning}",
+        tool_calls=[]
+    )
+    
+    # Create justification for blocking
+    justification = Justification(
+        step_number=state["current_step_index"] + 1,
+        reasoning=f"Step was blocked by safety guardrails. Risk level: {safety_status.risk_level}. Reason: {safety_status.reasoning}",
+        evidence=safety_status.blocked_reasons or [],
+        alternatives_considered=["Skip this step", "Abort entire task"]
+    )
+    state["justifications"].append(justification)
+    
+    # Mark status
+    state["status"] = "blocked"
+    
+    print(f"⚠️  Task blocked due to safety concerns")
+    
+    return state
+
+
+# ============================================================================
+# CONDITIONAL EDGES
+# ============================================================================
+
+def should_execute_or_refine(state: AgentState) -> Literal["execute", "refine"]:
+    """
+    Decide whether to execute or refine based on safety check.
+    
+    Args:
+        state: Current agent state
+        
+    Returns:
+        "execute" if safe, "refine" if unsafe
+    """
+    if state["safety_status"] and state["safety_status"].is_safe:
+        return "execute"
+    else:
+        return "refine"
+
+
+def should_continue_or_end(state: AgentState) -> Literal["continue", "end"]:
+    """
+    Decide whether to continue to next step or end the workflow.
+    
+    Args:
+        state: Current agent state
+        
+    Returns:
+        "continue" if more steps remain, "end" if done or blocked
+    """
+    # End if blocked
+    if state["safety_blocked"] and state["status"] == "blocked":
+        return "end"
+    
+    # End if error occurred
+    if state["error"]:
+        return "end"
+    
+    # End if all steps completed
+    if state["current_step_index"] >= state["plan"].total_steps:
+        return "end"
+    
+    # Continue to next step
+    return "continue"
+
+
+# ============================================================================
+# GRAPH CONSTRUCTION
+# ============================================================================
+
+def create_reasoning_graph() -> StateGraph:
+    """
+    Construct the full LangGraph state machine.
+    
+    Returns:
+        Compiled StateGraph ready for execution
+    """
+    # Create the graph
+    workflow = StateGraph(AgentState)
+    
+    # Add nodes
+    workflow.add_node("planner", planner_node)
+    workflow.add_node("safety", safety_node)
+    workflow.add_node("executor", executor_node)
+    workflow.add_node("logger", logger_node)
+    workflow.add_node("justification", justification_node)
+    workflow.add_node("iterator", step_iterator_node)
+    workflow.add_node("refiner", refiner_node)
+    
+    # Set entry point
+    workflow.set_entry_point("planner")
+    
+    # Add edges
+    workflow.add_edge("planner", "safety")
+    
+    # Conditional: safe -> execute, unsafe -> refine
+    workflow.add_conditional_edges(
+        "safety",
+        should_execute_or_refine,
+        {
+            "execute": "executor",
+            "refine": "refiner"
+        }
+    )
+    
+    # After execution: log -> justify -> iterate
+    workflow.add_edge("executor", "logger")
+    workflow.add_edge("logger", "justification")
+    workflow.add_edge("justification", "iterator")
+    
+    # After refining: go to iterator (to mark as done)
+    workflow.add_edge("refiner", "iterator")
+    
+    # Conditional: continue to next step or end
+    workflow.add_conditional_edges(
+        "iterator",
+        should_continue_or_end,
+        {
+            "continue": "safety",  # Loop back to safety check for next step
+            "end": END
+        }
+    )
+    
+    # Compile the graph
+    return workflow.compile()
+
+
+# ============================================================================
+# CONVENIENCE FUNCTION
+# ============================================================================
+
+def run_reasoning_task(task: str, task_id: str, user_id: str = None) -> AgentState:
+    """
+    Execute a reasoning task through the full pipeline.
+    
+    Args:
+        task: The task to solve
+        task_id: Unique identifier for this execution
+        user_id: Optional user identifier
+        
+    Returns:
+        Final agent state with results and logs
+    """
+    from state import create_initial_state
+    
+    # Create initial state
+    initial_state = create_initial_state(task, task_id, user_id)
+    
+    # Create and run the graph
+    graph = create_reasoning_graph()
+    
+    print(f"\n{'#'*60}")
+    print(f"🚀 STARTING REASONING PIPELINE")
+    print(f"   Task: {task}")
+    print(f"   Task ID: {task_id}")
+    print(f"{'#'*60}")
+    
+    # Execute
+    final_state = graph.invoke(initial_state)
+    
+    print(f"\n{'#'*60}")
+    print(f"🏁 REASONING PIPELINE COMPLETE")
+    print(f"   Status: {final_state['status']}")
+    print(f"   Steps Executed: {len(final_state['justifications'])}/{final_state['plan'].total_steps if final_state['plan'] else 0}")
+    print(f"   Cryptographic Logs: {len(final_state['logs'])}")
+    print(f"{'#'*60}\n")
+    
+    return final_state
+
+
+# ============================================================================
+# EXAMPLE USAGE
+# ============================================================================
+
+if __name__ == "__main__":
+    # Test the graph
+    result = run_reasoning_task(
+        task="Analyze the current state of AI safety research and provide 3 key findings",
+        task_id="test_001",
+        user_id="demo_user"
+    )
+    
+    # Print results
+    print("\n=== FINAL RESULTS ===")
+    print(f"Status: {result['status']}")
+    print(f"\nJustifications:")
+    for j in result['justifications']:
+        print(f"  Step {j.step_number}: {j.reasoning[:100]}...")
+    
+    print(f"\nCryptographic Audit Trail:")
+    for log in result['logs']:
+        print(f"  Step {log.step_number}: Hash {log.action_hash[:16]}... -> Root {log.merkle_root[:16]}...")

mock_tools.py

@@ -0,0 +1,221 @@
+"""
+Mock Crypto Tools for Secure Reasoning MCP Server
+Provides mock implementations for development and testing.
+Replace with real implementations when connecting to actual MCP server.
+"""
+import hashlib
+import json
+from datetime import datetime
+from typing import Optional, Dict, Any, List
+
+from schemas import (
+    HashRequest, HashResponse,
+    MerkleUpdateRequest, MerkleUpdateResponse,
+    WORMWriteRequest, WORMWriteResponse
+)
+
+
+class MockMerkleTree:
+    """
+    In-memory Merkle Tree for mock operations.
+    """
+    def __init__(self):
+        self.leaves: List[str] = []
+        self.root: Optional[str] = None
+    
+    def _calculate_root(self, leaves: List[str]) -> Optional[str]:
+        """Calculate Merkle root from a list of leaves."""
+        if not leaves:
+            return None
+        
+        if len(leaves) == 1:
+            return leaves[0]
+        
+        current_level = leaves[:]
+        
+        while len(current_level) > 1:
+            next_level = []
+            for i in range(0, len(current_level), 2):
+                left = current_level[i]
+                right = current_level[i + 1] if i + 1 < len(current_level) else left
+                combined = hashlib.sha256((left + right).encode()).hexdigest()
+                next_level.append(combined)
+            current_level = next_level
+        
+        return current_level[0]
+    
+    def update(self, new_hash: str) -> str:
+        """Add a new leaf and recalculate the Merkle root."""
+        self.leaves.append(new_hash)
+        self.root = self._calculate_root(self.leaves)
+        return self.root
+    
+    def get_proof(self, index: int) -> List[str]:
+        """Generate Merkle proof for a leaf at given index."""
+        if index < 0 or index >= len(self.leaves):
+            return []
+        
+        proof = []
+        current_index = index
+        current_level = self.leaves[:]
+        
+        while len(current_level) > 1:
+            is_right = current_index % 2 == 1
+            sibling_index = current_index - 1 if is_right else current_index + 1
+            
+            if sibling_index < len(current_level):
+                proof.append(current_level[sibling_index])
+            
+            current_index = current_index // 2
+            next_level = []
+            for i in range(0, len(current_level), 2):
+                left = current_level[i]
+                right = current_level[i + 1] if i + 1 < len(current_level) else left
+                combined = hashlib.sha256((left + right).encode()).hexdigest()
+                next_level.append(combined)
+            current_level = next_level
+        
+        return proof
+
+
+class MockCryptoTools:
+    """
+    Mock implementations of cryptographic tools.
+    Provides in-memory versions of hash, Merkle tree, and WORM storage.
+    """
+    
+    def __init__(self):
+        self.merkle_tree = MockMerkleTree()
+        self.worm_storage: Dict[str, Dict[str, Any]] = {}
+        self.storage_counter = 0
+    
+    def hash_tool(self, request: HashRequest) -> HashResponse:
+        """
+        Hash data using SHA-256 (or specified algorithm).
+        
+        Args:
+            request: HashRequest with data and algorithm
+            
+        Returns:
+            HashResponse with the hash digest
+        """
+        data = request.data
+        
+        # Ensure deterministic serialization
+        if isinstance(data, dict):
+            data_str = json.dumps(data, sort_keys=True)
+        else:
+            data_str = str(data)
+        
+        # Compute hash (only SHA-256 implemented for mock)
+        hash_value = hashlib.sha256(data_str.encode()).hexdigest()
+        
+        return HashResponse(
+            hash=hash_value,
+            algorithm=request.algorithm,
+            timestamp=datetime.utcnow()
+        )
+    
+    def merkle_update_tool(self, request: MerkleUpdateRequest) -> MerkleUpdateResponse:
+        """
+        Add a hash to the Merkle tree and return updated root.
+        
+        Args:
+            request: MerkleUpdateRequest with leaf hash
+            
+        Returns:
+            MerkleUpdateResponse with new root and proof
+        """
+        # Add leaf and get new root
+        new_root = self.merkle_tree.update(request.leaf_hash)
+        leaf_index = len(self.merkle_tree.leaves) - 1
+        
+        # Generate proof for the new leaf
+        proof = self.merkle_tree.get_proof(leaf_index)
+        
+        return MerkleUpdateResponse(
+            merkle_root=new_root,
+            leaf_index=leaf_index,
+            proof=proof,
+            tree_size=len(self.merkle_tree.leaves)
+        )
+    
+    def worm_write_tool(self, request: WORMWriteRequest) -> WORMWriteResponse:
+        """
+        Write data to WORM (Write Once, Read Many) storage.
+        
+        Args:
+            request: WORMWriteRequest with entry ID and data
+            
+        Returns:
+            WORMWriteResponse with storage confirmation
+        """
+        # Check if entry already exists (WORM = no overwrites)
+        if request.entry_id in self.worm_storage:
+            return WORMWriteResponse(
+                success=False,
+                storage_path=f"worm://{request.entry_id}",
+                verification_hash="",
+                timestamp=datetime.utcnow()
+            )
+        
+        # Store the data
+        self.storage_counter += 1
+        storage_path = f"worm://mock/{self.storage_counter}/{request.entry_id}"
+        
+        # Compute verification hash
+        verification_hash = hashlib.sha256(
+            json.dumps(request.data, sort_keys=True).encode()
+        ).hexdigest()
+        
+        # Store immutably
+        self.worm_storage[request.entry_id] = {
+            "data": request.data,
+            "merkle_root": request.merkle_root,
+            "verification_hash": verification_hash,
+            "storage_path": storage_path,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        
+        return WORMWriteResponse(
+            success=True,
+            storage_path=storage_path,
+            verification_hash=verification_hash,
+            timestamp=datetime.utcnow()
+        )
+    
+    def worm_read_tool(self, entry_id: str) -> Optional[Dict[str, Any]]:
+        """
+        Read data from WORM storage (for verification).
+        
+        Args:
+            entry_id: The ID of the entry to read
+            
+        Returns:
+            The stored data or None if not found
+        """
+        return self.worm_storage.get(entry_id)
+    
+    def verify_proof(self, target_hash: str, proof: List[str], root: str) -> bool:
+        """
+        Verify a Merkle proof.
+        
+        Args:
+            target_hash: The hash to verify
+            proof: List of sibling hashes
+            root: Expected Merkle root
+            
+        Returns:
+            True if proof is valid, False otherwise
+        """
+        current_hash = target_hash
+        
+        for sibling_hash in proof:
+            # Combine in lexicographic order for consistency
+            if current_hash < sibling_hash:
+                combined = current_hash + sibling_hash
+            else:
+                combined = sibling_hash + current_hash
+            current_hash = hashlib.sha256(combined.encode()).hexdigest()
+        
+        return current_hash == root

prompts.py

@@ -0,0 +1,407 @@
+"""
+Prompt Templates for Secure Reasoning MCP Server
+Optimized for Claude 3.5 Sonnet with strict JSON output requirements.
+"""
+
+# ============================================================================
+# PLANNER PROMPT
+# ============================================================================
+
+PLANNER_SYSTEM_PROMPT = """You are a strategic planning agent for a secure reasoning system. Your role is to break down complex tasks into clear, executable steps.
+
+**Your Responsibilities:**
+1. Analyze the user's task thoroughly
+2. Create a step-by-step execution plan
+3. Identify which steps require external tools
+4. Ensure steps are atomic (one clear action per step)
+5. Order steps logically with dependencies respected
+
+**Output Format:**
+You MUST respond with valid JSON only, no preamble or explanation. Use this exact structure:
+
+{
+  "steps": [
+    {
+      "step_number": 1,
+      "action": "Clear description of what to do",
+      "expected_outcome": "What this step should achieve",
+      "requires_tools": true
+    }
+  ],
+  "total_steps": 3
+}
+
+**Guidelines:**
+- Break complex tasks into 3-7 steps (avoid over-fragmenting)
+- Each step should be verifiable and have a clear completion criterion
+- Mark `requires_tools: true` if the step needs data retrieval, computation, or external API calls
+- Mark `requires_tools: false` for pure reasoning, analysis, or synthesis steps
+- Consider safety: avoid steps that could cause harm or violate ethics
+- Be specific: "Search for X" not "Find information"
+
+**Examples of Good Steps:**
+✓ "Search web for current S&P 500 price and extract the value"
+✓ "Analyze the retrieved data to identify the trend over last 30 days"
+✓ "Synthesize findings into a 3-sentence summary for the user"
+
+**Examples of Bad Steps:**
+✗ "Do research" (too vague)
+✗ "Get data and analyze it" (multiple actions)
+✗ "Figure out the answer" (not actionable)"""
+
+PLANNER_USER_TEMPLATE = """Task: {task}
+
+Please generate a detailed execution plan for this task. Return ONLY valid JSON, nothing else."""
+
+
+# ============================================================================
+# SAFETY GUARDRAIL PROMPT
+# ============================================================================
+
+SAFETY_SYSTEM_PROMPT = """You are a safety validation system for an AI reasoning agent. Your role is to prevent harmful, unethical, or dangerous actions BEFORE they are executed.
+
+**Your Responsibilities:**
+1. Evaluate if a proposed action is safe to execute
+2. Assess potential risks across multiple dimensions
+3. Block actions that could cause harm
+4. Provide clear reasoning for safety decisions
+
+**Risk Assessment Dimensions:**
+- **Harm to People**: Physical, psychological, or financial harm
+- **Privacy Violations**: Unauthorized data access or surveillance
+- **Security Risks**: Hacking, exploits, or system compromise
+- **Misinformation**: Generating or spreading false information
+- **Illegal Activities**: Actions that violate laws or regulations
+- **Ethical Concerns**: Manipulation, deception, or unfair bias
+- **Discrimination**: Actions that target protected groups
+
+**Output Format:**
+You MUST respond with valid JSON only. Use this exact structure:
+
+{
+  "is_safe": true,
+  "risk_level": "safe",
+  "reasoning": "Detailed explanation of the safety assessment",
+  "blocked_reasons": []
+}
+
+**Risk Levels:**
+- `"safe"`: No concerns, proceed with execution
+- `"low"`: Minor concerns but acceptable with monitoring
+- `"medium"`: Moderate risk, requires careful handling
+- `"high"`: Significant risk, should be avoided
+- `"critical"`: Severe risk, must be blocked immediately
+
+**Decision Rules:**
+- If `risk_level` is "high" or "critical", set `is_safe: false`
+- If `risk_level` is "medium", use judgment based on context
+- If `risk_level` is "safe" or "low", set `is_safe: true`
+- Always populate `blocked_reasons` array when `is_safe: false`
+
+**Approval Bias:**
+Default to allowing actions unless there is a clear safety concern. The system should enable legitimate tasks while protecting against harm.
+
+**Examples:**
+
+✓ SAFE: "Search for Python documentation on list comprehensions"
+✓ SAFE: "Calculate the average of [1, 2, 3, 4, 5]"
+✓ SAFE: "Summarize the key findings from the retrieved article"
+
+✗ UNSAFE: "Generate a phishing email template"
+✗ UNSAFE: "Find personal contact information for [person]"
+✗ UNSAFE: "Create instructions for synthesizing harmful substances"
+✗ UNSAFE: "Search for ways to bypass security systems"
+"""
+
+SAFETY_USER_TEMPLATE = """**Step to Validate:**
+{step_description}
+
+**Full Context:**
+- Task: {task}
+- Step Number: {step_number} of {total_steps}
+- Previous Steps: {previous_steps}
+
+**Additional Context:**
+{additional_context}
+
+Please evaluate if this step is safe to execute. Return ONLY valid JSON."""
+
+
+# ============================================================================
+# EXECUTOR/ROUTER PROMPT
+# ============================================================================
+
+EXECUTOR_SYSTEM_PROMPT = """You are an intelligent action executor for a reasoning system. Your role is to execute approved steps and determine which tools (if any) are needed.
+
+**Available Tools:**
+1. **web_search**: Search the internet for current information
+2. **web_fetch**: Retrieve full content from a specific URL
+3. **calculate**: Perform mathematical computations
+4. **code_execute**: Run Python code in a sandbox
+5. **internal_reasoning**: Use pure reasoning without external tools
+
+**Your Responsibilities:**
+1. Determine which tool best accomplishes the step
+2. Extract the specific parameters needed for the tool
+3. Execute the action or call the appropriate tool
+4. Return structured results
+
+**Output Format:**
+You MUST respond with valid JSON only:
+
+{
+  "tool_needed": "web_search",
+  "tool_params": {
+    "query": "specific search query"
+  },
+  "reasoning": "Why this tool was selected"
+}
+
+OR if no external tool is needed:
+
+{
+  "tool_needed": "internal_reasoning",
+  "tool_params": null,
+  "reasoning": "This can be solved through analysis alone",
+  "direct_result": "The answer or analysis"
+}
+
+**Tool Selection Guidelines:**
+- Use `web_search` for: current events, real-time data, factual lookups
+- Use `web_fetch` for: retrieving specific documents or web pages
+- Use `calculate` for: mathematical operations, data analysis
+- Use `code_execute` for: complex computations, data transformations
+- Use `internal_reasoning` for: analysis, synthesis, planning, summarization
+
+**Important:**
+- Choose the MINIMAL tool necessary (don't over-engineer)
+- Be specific with parameters (exact search terms, precise calculations)
+- If a step can be done without tools, use `internal_reasoning`"""
+
+EXECUTOR_USER_TEMPLATE = """**Step to Execute:**
+{step_description}
+
+**Context:**
+- Task: {task}
+- Expected Outcome: {expected_outcome}
+- Requires Tools: {requires_tools}
+- Previous Results: {previous_results}
+
+Determine how to execute this step and return the appropriate JSON structure."""
+
+
+# ============================================================================
+# JUSTIFICATION PROMPT
+# ============================================================================
+
+JUSTIFICATION_SYSTEM_PROMPT = """You are a transparency and explainability agent. Your role is to explain WHY actions were taken in clear, understandable language.
+
+**Your Responsibilities:**
+1. Explain the reasoning behind the executed action
+2. Connect the action to the overall task goal
+3. Cite specific evidence or data that informed the decision
+4. Note any alternative approaches that were considered
+5. Make the reasoning transparent and auditable
+
+**Output Format:**
+You MUST respond with valid JSON only:
+
+{
+  "step_number": 1,
+  "reasoning": "Clear natural language explanation of why this action was taken",
+  "evidence": [
+    "Specific fact or data point that supported this decision",
+    "Another supporting piece of evidence"
+  ],
+  "alternatives_considered": [
+    "Alternative approach 1 and why it wasn't chosen",
+    "Alternative approach 2 and why it wasn't chosen"
+  ]
+}
+
+**Explanation Guidelines:**
+- Write for a technical but non-expert audience
+- Be specific: cite actual data, tool outputs, or reasoning steps
+- Connect each action to the broader task goal
+- Acknowledge uncertainty when present
+- Explain trade-offs in the decision-making process
+
+**Good Justifications:**
+✓ "Used web_search because the task requires current S&P 500 price (data changes daily). Retrieved price of $6,852.34 from reliable financial source. Alternative of using cached data was rejected due to staleness risk."
+
+✓ "Applied internal_reasoning to synthesize findings because the step requires analysis of existing data, not new information retrieval. Combined results from steps 1-3 to identify the trend pattern. Alternative of using code_execute would be over-engineering for this simple synthesis task."
+
+**Bad Justifications:**
+✗ "Performed the action." (no explanation)
+✗ "It seemed like the right thing to do." (vague)
+✗ "The system told me to." (not transparent)"""
+
+JUSTIFICATION_USER_TEMPLATE = """**Action Taken:**
+- Step: {step_description}
+- Tool Used: {tool_used}
+- Result: {execution_result}
+
+**Context:**
+- Task: {task}
+- Step Number: {step_number} of {total_steps}
+- Expected Outcome: {expected_outcome}
+
+Please provide a clear justification for why this action was taken and how it advances the task. Return ONLY valid JSON."""
+
+
+# ============================================================================
+# FINAL SYNTHESIS PROMPT
+# ============================================================================
+
+SYNTHESIS_SYSTEM_PROMPT = """You are a final synthesis agent. Your role is to compile all executed steps into a coherent final answer for the user.
+
+**Your Responsibilities:**
+1. Review all executed steps and their results
+2. Synthesize findings into a clear, complete answer
+3. Ensure the answer directly addresses the original task
+4. Include relevant evidence and data
+5. Maintain appropriate confidence levels
+
+**Output Format:**
+Return a natural language response (NOT JSON for this prompt). Structure your answer as:
+
+1. **Direct Answer**: Lead with the answer to the task
+2. **Supporting Evidence**: Key data or findings that support the answer
+3. **Confidence Level**: Your certainty in this answer (high/medium/low)
+4. **Caveats**: Any limitations or uncertainties
+
+**Quality Guidelines:**
+- Be concise but complete
+- Cite specific data from the execution steps
+- Acknowledge uncertainty where present
+- Use clear, accessible language
+- Ensure the answer is actionable"""
+
+SYNTHESIS_USER_TEMPLATE = """**Original Task:**
+{task}
+
+**Executed Steps Summary:**
+{steps_summary}
+
+**Results from Each Step:**
+{all_results}
+
+Please synthesize these findings into a final answer for the user."""
+
+
+# ============================================================================
+# ERROR HANDLING PROMPTS
+# ============================================================================
+
+ERROR_ANALYSIS_PROMPT = """You are an error analysis agent. A step in the reasoning chain has failed.
+
+**Your Task:**
+Analyze the error and determine:
+1. What went wrong
+2. Whether the error is recoverable
+3. What corrective action should be taken
+
+**Output JSON:**
+{
+  "error_type": "tool_failure|validation_error|safety_block|timeout",
+  "is_recoverable": true,
+  "suggested_action": "retry|skip|abort|modify_step",
+  "explanation": "Clear explanation of the error and recommendation"
+}
+
+**Error Details:**
+Step: {step_description}
+Error: {error_message}
+Context: {context}
+
+Return ONLY valid JSON."""
+
+
+# ============================================================================
+# HELPER FUNCTIONS FOR PROMPT FORMATTING
+# ============================================================================
+
+def format_planner_prompt(task: str) -> dict:
+    """Format the planner prompt with task context."""
+    return {
+        "system": PLANNER_SYSTEM_PROMPT,
+        "user": PLANNER_USER_TEMPLATE.format(task=task)
+    }
+
+
+def format_safety_prompt(
+    step_description: str,
+    task: str,
+    step_number: int,
+    total_steps: int,
+    previous_steps: str = "None",
+    additional_context: str = "None"
+) -> dict:
+    """Format the safety validation prompt."""
+    return {
+        "system": SAFETY_SYSTEM_PROMPT,
+        "user": SAFETY_USER_TEMPLATE.format(
+            step_description=step_description,
+            task=task,
+            step_number=step_number,
+            total_steps=total_steps,
+            previous_steps=previous_steps,
+            additional_context=additional_context
+        )
+    }
+
+
+def format_executor_prompt(
+    step_description: str,
+    task: str,
+    expected_outcome: str,
+    requires_tools: bool,
+    previous_results: str = "None"
+) -> dict:
+    """Format the executor/router prompt."""
+    return {
+        "system": EXECUTOR_SYSTEM_PROMPT,
+        "user": EXECUTOR_USER_TEMPLATE.format(
+            step_description=step_description,
+            task=task,
+            expected_outcome=expected_outcome,
+            requires_tools=requires_tools,
+            previous_results=previous_results
+        )
+    }
+
+
+def format_justification_prompt(
+    step_description: str,
+    tool_used: str,
+    execution_result: str,
+    task: str,
+    step_number: int,
+    total_steps: int,
+    expected_outcome: str
+) -> dict:
+    """Format the justification prompt."""
+    return {
+        "system": JUSTIFICATION_SYSTEM_PROMPT,
+        "user": JUSTIFICATION_USER_TEMPLATE.format(
+            step_description=step_description,
+            tool_used=tool_used,
+            execution_result=execution_result,
+            task=task,
+            step_number=step_number,
+            total_steps=total_steps,
+            expected_outcome=expected_outcome
+        )
+    }
+
+
+def format_synthesis_prompt(task: str, steps_summary: str, all_results: str) -> dict:
+    """Format the final synthesis prompt."""
+    return {
+        "system": SYNTHESIS_SYSTEM_PROMPT,
+        "user": SYNTHESIS_USER_TEMPLATE.format(
+            task=task,
+            steps_summary=steps_summary,
+            all_results=all_results
+        )
+    }

requirements.txt

@@ -0,0 +1,18 @@
+# Core dependencies
+gradio>=5.0.0
+python-dotenv>=1.0.0
+
+# LangChain / LangGraph
+langgraph>=0.2.0
+langchain>=0.3.0
+langchain-core>=0.3.0
+langchain-anthropic>=0.2.0
+
+# Data validation
+pydantic>=2.0.0
+
+# MCP Server
+fastmcp>=0.1.0
+
+# Utilities
+httpx>=0.25.0

schemas.py

@@ -0,0 +1,135 @@
+"""
+Pydantic Models for Secure Reasoning MCP Server
+Defines all input/output schemas for the agent and tool interfaces.
+"""
+from typing import List, Optional, Dict, Any, Literal
+from pydantic import BaseModel, Field
+from datetime import datetime
+
+
+# ============================================================================
+# AGENT INPUT/OUTPUT MODELS
+# ============================================================================
+
+class TaskRequest(BaseModel):
+    """External API request to start a reasoning task."""
+    task: str = Field(..., description="The task/query for the agent to solve")
+    user_id: Optional[str] = Field(None, description="Optional user identifier for audit trail")
+    metadata: Optional[Dict[str, Any]] = Field(default_factory=dict, description="Additional context")
+
+
+class StepPlan(BaseModel):
+    """A single step in the agent's execution plan."""
+    step_number: int = Field(..., description="Sequential step index")
+    action: str = Field(..., description="What action to take")
+    expected_outcome: str = Field(..., description="What this step should achieve")
+    requires_tools: bool = Field(False, description="Whether this step needs tool execution")
+
+
+class ExecutionPlan(BaseModel):
+    """The full plan generated by the agent."""
+    steps: List[StepPlan] = Field(..., description="Ordered list of steps")
+    total_steps: int = Field(..., description="Total number of steps")
+    generated_at: datetime = Field(default_factory=datetime.utcnow)
+
+
+class SafetyCheckResult(BaseModel):
+    """Result from the safety validation LLM."""
+    is_safe: bool = Field(..., description="Whether the step is approved")
+    risk_level: Literal["safe", "low", "medium", "high", "critical"] = Field(..., description="Risk assessment")
+    reasoning: str = Field(..., description="Explanation of the safety decision")
+    blocked_reasons: Optional[List[str]] = Field(None, description="Specific safety violations if blocked")
+
+
+class ExecutionResult(BaseModel):
+    """Result from executing a single step."""
+    success: bool = Field(..., description="Whether execution succeeded")
+    output: Any = Field(None, description="The result data")
+    error: Optional[str] = Field(None, description="Error message if failed")
+    tool_calls: List[str] = Field(default_factory=list, description="Tools that were invoked")
+
+
+class Justification(BaseModel):
+    """Agent's explanation for why it took an action."""
+    step_number: int
+    reasoning: str = Field(..., description="Natural language explanation")
+    evidence: Optional[List[str]] = Field(None, description="Supporting facts or data")
+    alternatives_considered: Optional[List[str]] = Field(None, description="Other approaches considered")
+
+
+class TaskResponse(BaseModel):
+    """Final response returned to the user."""
+    task_id: str = Field(..., description="Unique identifier for this execution")
+    status: Literal["completed", "failed", "blocked"] = Field(..., description="Final status")
+    result: Optional[Any] = Field(None, description="The final answer or output")
+    plan: ExecutionPlan = Field(..., description="The plan that was executed")
+    justifications: List[Justification] = Field(..., description="Reasoning for each step")
+    logs: List["CryptoLogEntry"] = Field(..., description="Cryptographic audit trail")
+    error: Optional[str] = Field(None, description="Error message if failed")
+
+
+# ============================================================================
+# CRYPTOGRAPHIC TOOL INTERFACES (for teammate's implementations)
+# ============================================================================
+
+class HashRequest(BaseModel):
+    """Request to hash data."""
+    data: str = Field(..., description="Data to hash (JSON string or plain text)")
+    algorithm: Literal["sha256", "sha3_256", "blake2b"] = Field("sha256", description="Hash algorithm")
+
+
+class HashResponse(BaseModel):
+    """Response from hash tool."""
+    hash: str = Field(..., description="Hexadecimal hash digest")
+    algorithm: str = Field(..., description="Algorithm used")
+    timestamp: datetime = Field(default_factory=datetime.utcnow)
+
+
+class MerkleUpdateRequest(BaseModel):
+    """Request to add a hash to the Merkle tree."""
+    leaf_hash: str = Field(..., description="Hash to add as a new leaf")
+    metadata: Optional[Dict[str, Any]] = Field(None, description="Additional context to store")
+
+
+class MerkleUpdateResponse(BaseModel):
+    """Response from Merkle tree update."""
+    merkle_root: str = Field(..., description="Updated Merkle root hash")
+    leaf_index: int = Field(..., description="Index of the new leaf")
+    proof: List[str] = Field(..., description="Merkle proof path")
+    tree_size: int = Field(..., description="Total leaves in tree")
+
+
+class WORMWriteRequest(BaseModel):
+    """Request to write immutable data to WORM storage."""
+    entry_id: str = Field(..., description="Unique identifier for this entry")
+    data: Dict[str, Any] = Field(..., description="Data to store permanently")
+    merkle_root: str = Field(..., description="Current Merkle root for verification")
+
+
+class WORMWriteResponse(BaseModel):
+    """Response from WORM storage write."""
+    success: bool = Field(..., description="Whether write succeeded")
+    storage_path: str = Field(..., description="Where data was stored")
+    verification_hash: str = Field(..., description="Hash of the stored data for verification")
+    timestamp: datetime = Field(default_factory=datetime.utcnow)
+
+
+class CryptoLogEntry(BaseModel):
+    """A single entry in the cryptographic audit trail."""
+    step_number: int
+    action_hash: str = Field(..., description="Hash of the action taken")
+    merkle_root: str = Field(..., description="Merkle root after this action")
+    worm_path: Optional[str] = Field(None, description="WORM storage location")
+    timestamp: datetime = Field(default_factory=datetime.utcnow)
+
+
+# ============================================================================
+# ERROR MODELS
+# ============================================================================
+
+class ErrorResponse(BaseModel):
+    """Standard error response."""
+    error: str = Field(..., description="Error message")
+    error_type: str = Field(..., description="Category of error")
+    details: Optional[Dict[str, Any]] = Field(None, description="Additional error context")
+    timestamp: datetime = Field(default_factory=datetime.utcnow)

server.py

@@ -0,0 +1,60 @@
+from fastmcp import FastMCP
+from crypto_engine import hash_tool, worm_write_tool, proof_generate_tool, verify_proof_tool
+import json
+
+# Initialize the MCP server
+mcp = FastMCP("Secure Reasoning Server")
+
+
+@mcp.tool()
+def hash_data(data: str) -> str:
+    """
+    Hash a string or JSON data using SHA-256.
+    Input: data (string or JSON-serializable object as string)
+    Output: SHA-256 hex digest
+    """
+    return hash_tool(data)
+
+
+@mcp.tool()
+def write_to_worm(step_data: str, hash_value: str, merkle_root: str) -> str:
+    """
+    Write a step record to WORM (Write Once, Read Many) storage.
+    Input: step_data (JSON string), hash_value (hex string), merkle_root (hex string)
+    Output: JSON record with id, timestamp, step, hash, and root
+    """
+    step_dict = json.loads(step_data) if isinstance(step_data, str) else step_data
+    record = worm_write_tool(step_dict, hash_value, merkle_root)
+    return json.dumps(record)
+
+
+@mcp.tool()
+def generate_proof(record_id: int) -> str:
+    """
+    Generate a Merkle proof for a specific record in the WORM log.
+    Input: record_id (integer, the ID of the record)
+    Output: JSON containing record_id, hash, merkle_proof, merkle_root, timestamp, and step_details
+    """
+    proof = proof_generate_tool(record_id)
+    if proof is None:
+        return json.dumps({"error": f"Record with ID {record_id} not found"})
+    return json.dumps(proof)
+
+
+@mcp.tool()
+def verify_proof(target_hash: str, merkle_proof: str, merkle_root: str) -> str:
+    """
+    Verify if a target_hash belongs to the merkle_root using the merkle_proof.
+    Input: target_hash (hex string), merkle_proof (JSON string of proof array), merkle_root (hex string)
+    Output: JSON with result (true/false) and verification status message
+    """
+    proof_list = json.loads(merkle_proof) if isinstance(merkle_proof, str) else merkle_proof
+    is_valid = verify_proof_tool(target_hash, proof_list, merkle_root)
+    return json.dumps({
+        "verified": is_valid,
+        "message": "Proof verified successfully" if is_valid else "Proof verification failed - possible tampering"
+    })
+
+
+if __name__ == "__main__":
+    mcp.run()

state.py

@@ -0,0 +1,113 @@
+"""
+LangGraph State Definition for Secure Reasoning Agent
+Tracks all state through the Chain-of-Checks workflow.
+"""
+from typing import TypedDict, List, Optional, Annotated
+from operator import add
+from langchain_core.messages import BaseMessage
+
+from schemas import (
+    ExecutionPlan,
+    SafetyCheckResult,
+    CryptoLogEntry,
+    Justification,
+    ExecutionResult
+)
+
+
+class AgentState(TypedDict):
+    """
+    State tracked throughout the LangGraph execution.
+    
+    The state flows through: Plan → Safety Check → Execute → Log → Justify → Loop
+    """
+    
+    # ========================================================================
+    # CONVERSATION & CONTEXT
+    # ========================================================================
+    messages: Annotated[List[BaseMessage], add]
+    """Chat history with the user and internal LLM calls. Uses 'add' reducer to append."""
+    
+    task: str
+    """The original user task/query."""
+    
+    task_id: str
+    """Unique identifier for this execution (for audit trail)."""
+    
+    user_id: Optional[str]
+    """Optional user identifier for multi-user environments."""
+    
+    # ========================================================================
+    # PLANNING STATE
+    # ========================================================================
+    plan: Optional[ExecutionPlan]
+    """The generated execution plan with all steps."""
+    
+    current_step_index: int
+    """Which step we're currently processing (0-indexed)."""
+    
+    # ========================================================================
+    # SAFETY & VALIDATION
+    # ========================================================================
+    safety_status: Optional[SafetyCheckResult]
+    """Result of safety check for current step. None if not yet checked."""
+    
+    safety_blocked: bool
+    """Quick flag: True if any step was blocked by safety guardrails."""
+    
+    # ========================================================================
+    # EXECUTION STATE
+    # ========================================================================
+    execution_result: Optional[ExecutionResult]
+    """Result from executing the current step."""
+    
+    final_result: Optional[str]
+    """The final answer/output when all steps complete."""
+    
+    # ========================================================================
+    # AUDIT TRAIL & CRYPTOGRAPHIC LOGGING
+    # ========================================================================
+    logs: List[CryptoLogEntry]
+    """Cryptographic proofs for each executed step (Merkle roots, hashes, etc.)."""
+    
+    justifications: List[Justification]
+    """Agent's reasoning for each action taken."""
+    
+    # ========================================================================
+    # ERROR HANDLING
+    # ========================================================================
+    error: Optional[str]
+    """Error message if execution fails."""
+    
+    status: str
+    """Current execution status: 'planning', 'executing', 'completed', 'failed', 'blocked'."""
+
+
+def create_initial_state(task: str, task_id: str, user_id: Optional[str] = None) -> AgentState:
+    """
+    Factory function to create a fresh AgentState for a new task.
+    
+    Args:
+        task: The user's task/query
+        task_id: Unique identifier for this execution
+        user_id: Optional user identifier
+        
+    Returns:
+        Initialized AgentState ready for LangGraph processing
+    """
+    return AgentState(
+        messages=[],
+        task=task,
+        task_id=task_id,
+        user_id=user_id,
+        plan=None,
+        current_step_index=0,
+        safety_status=None,
+        safety_blocked=False,
+        execution_result=None,
+        final_result=None,
+        logs=[],
+        justifications=[],
+        error=None,
+        status="planning"
+    )