refactor: add parse_mcp_output into utils, give prompt_temp as a resource of server, improve prompt

postgre_mcp_client.py

@@ -7,11 +7,12 @@ from langgraph.prebuilt import create_react_agent
 from langchain.chat_models import init_chat_model
 from conversation_memory import ConversationMemory
 
+from utils import parse_mcp_output
+
 llm = init_chat_model(model="gemini-2.0-flash-lite", model_provider="google_genai",api_key ="AIzaSyAuxYmci0DVU5l5L_YcxLlxHzR5MLn70js")
 
 server_params = StdioServerParameters(
     command="python",
-    # buraya full path konulmalı
     args=[r"C:\Users\yukcus\Desktop\MCPTest\postgre_mcp_server.py"],
 )
 
@@ -20,81 +21,6 @@ The users table stores information about the individuals who use the application
 
 The posts table represents content created by users, such as blog posts or messages. Like the users table, each entry has a unique, auto-incrementing id as the primary key. The user_id field links each post to its author by referencing the id field in the users table, establishing a one-to-many relationship between users and posts. The title column holds a brief summary or headline of the post, while the content field contains the full text. A created_at timestamp is also included to record when each post was created, with a default value of the current time.
 """
-prompt_temp = ""
-
-def parse_mcp_output(output_dict):
-    result = []
-    messages = output_dict.get("messages", [])
-
-    for msg in messages:
-        role_name = msg.__class__.__name__  # Example: HumanMessage, AIMessage, ToolMessage
-        content = getattr(msg, "content", "")
-
-        # AIMessage with tool call
-        if role_name == "AIMessage":
-            function_call = getattr(msg, "additional_kwargs", {}).get("function_call")
-            if function_call:
-                tool_name = function_call.get("name")
-                arguments = function_call.get("arguments")
-
-                # Check if arguments is a JSON string or a dict
-                if isinstance(arguments, str):
-                    import json
-                    try:
-                        arguments_dict = json.loads(arguments)
-                    except json.JSONDecodeError:
-                        arguments_dict = {}
-                else:
-                    arguments_dict = arguments or {}
-
-                # Check for presence of "query" key
-                if "query" in arguments_dict:
-                    print("query detected!!!")
-                    print(f"ai said:{content[0]}")
-                    print(arguments_dict["query"])
-
-                    result.append({
-                        "type": "ai_function_call",
-                        "ai_said": content,
-                        "tool": tool_name,
-                        "args": arguments
-                    })
-                else:
-                    print(f"ai said:{content}")
-                    result.append({
-                        "type": "ai_function_call",
-                        "ai_said": content,
-                        "tool": tool_name,
-                        "args": arguments
-                    })
-
-            else:
-                print(f"ai final answer:{content}")
-                result.append({
-                    "type": "ai_final_answer",
-                    "ai_said": content
-                })
-
-        # ToolMessage
-        elif role_name == "ToolMessage":
-            tool_name = getattr(msg, "name", None)
-            print(f"tool response:{content}")
-            result.append({
-                "type": "tool_response",
-                "tool": tool_name,
-                "response": content
-            })
-
-        elif role_name == "HumanMessage":
-            result.append({
-                "type": "user_message",
-                "content": content
-            })
-
-
-    return result
-
-
 
 request = "can you show me the result of the join of all tables?"
 request2 = "how many columns are there in this joined table?"
@@ -105,11 +31,6 @@ async def main():
             await session.initialize()
 
             memory = ConversationMemory()
-            prompt = ""
-            with open(r"C:\Users\yukcus\Desktop\MCPTest\prompt_temp.txt", 'r', encoding='utf-8') as file:
-                prompt_temp = file.read()
-
-            prompt += prompt_temp
 
             # Get tools
             tools = await load_mcp_tools(session)
@@ -122,12 +43,21 @@ async def main():
             past_results = memory.get_last_n_results()
             past_requests = memory.get_all_user_messages()
 
-            prompt = prompt.format(
+            uri = f"resource://base_prompt_table"
+            resource = await session.read_resource(uri)
+            base_prompt = resource.contents[0].text
+
+            # Create a formatted string of tools
+            tools_str = "\n".join([f"- {tool.name}: {tool.description}" for tool in tools])
+
+
+            prompt = base_prompt.format(
                 user_requests=past_requests,
                 past_tools=past_tools,
                 last_queries=past_queries,
                 last_results=past_results,
-                new_request = request2
+                new_request = request2,
+                tools = tools_str
             )
 
 
@@ -137,7 +67,7 @@ async def main():
             agent_response = await agent.ainvoke({"messages": prompt})
 
 
-            parsed_steps = parse_mcp_output(agent_response)
+            parsed_steps, query_store = parse_mcp_output(agent_response)
 
 
             memory.update_from_parsed(parsed_steps)

postgre_mcp_server.py

@@ -49,6 +49,161 @@ mcp = FastMCP(
 )
 
 
+@mcp.resource(
+    uri="resource://base_prompt_table",
+    name="base_prompt_table",
+    description="A base prompt to generate description of a table"
+)
+async def base_prompt_table() -> str:
+    """Returns a base prompt to generate description of a table"""
+
+    base_prompt = """
+
+                    ==========================
+                    # Your Role
+                    ==========================
+
+                    You are an expert in generating SQL queries and interacting with a PostgreSQL database using **FastMCP tools**. These tools allow you to:
+
+                    - List available tables
+                    - Retrieve schema details
+                    - Execute SQL queries
+
+                    Each tool may also return previews or summaries of table contents to help you better understand the data structure.
+
+                    You also have access to **short-term memory**, which stores relevant context from earlier queries. If memory contains the needed information, you **must use it** instead of repeating tool calls with the same input. Avoid redundant tool usage unless:
+                    - The memory is empty, or
+                    - A tool's output is outdated or missing
+
+                    ---
+
+                    ==========================
+                    # Your Objective
+                    ==========================
+
+                    When a user submits a request, follow these steps:
+
+                    1. Analyze the request to determine the desired data or action.
+                    2. Use tools to gather any necessary information (e.g., list tables, get schema).
+                    3. Generate a valid SQL query (such as **SELECT**, **COUNT**, or other read-only operations) and clearly display the full query.
+                    4. Execute the query and return the result.
+                    5. Chain tools logically to build toward the answer.
+                    6. Explain your reasoning at every step for clarity and transparency.
+
+                    ---
+
+                    ==========================
+                    # Critical Rules
+                    ==========================
+
+                    - Only use **read-only** SQL queries such as **SELECT**, **COUNT**, or queries with **GROUP BY**, **ORDER BY**, etc.
+                    - **Never** use destructive operations like **DELETE**, **UPDATE**, **INSERT**, or **DROP**.
+                    - Always show the SQL query you generate along with the execution result.
+                    - Validate SQL syntax before execution.
+                    - Never assume table or column names. Use tools to confirm structure.
+                    - Use memory efficiently. Don’t rerun a tool unless necessary.
+
+                    ---
+
+                    ==========================
+                    # Short-Term Memory
+                    ==========================
+
+                    You have access to the following memory from this conversation. Use it if applicable for the current request.
+
+                    - Previous user requests: {user_requests}
+                    - Tools used so far: {past_tools}
+                    - Last SQL queries: {last_queries}
+                    - Last result preview: {last_results}
+
+                    ---
+
+                    ==========================
+                    # Tools
+                    ==========================
+
+                    You can use the following FastMCP tools. These allow you to create **read-only** queries, such as `SELECT`, `COUNT`, or queries with `GROUP BY`, `ORDER BY`, and similar clauses. You may chain tools together to gather the necessary information before generating your SQL query.
+
+                    {tools}
+
+                    ---
+
+                    ==========================
+                    # Tool Usage Examples
+                    ==========================
+
+                    ### Example 1 — Unknown Table Name:
+                    **User Request:** "Get the total sales for each product."
+                    **Steps:**
+                    1. List Tables → Identify a table like `sales_data`.
+                    2. Get Schema for `sales_data` → Confirm columns like `product_name`, `total_sales`.
+                    3. Generate and execute query:
+                       ```sql
+                       SELECT product_name, SUM(total_sales) FROM sales_data GROUP BY product_name;
+
+                    ### Example 2 — Schema Uncertainty:
+                    **User Request:** "Show customer emails from the database."
+
+                    **Steps:**
+                    1. Use memory to check if we already retrieved schema.
+                    2. If not, List Tables → Identify a table like `customers`.
+                    3. Get Schema for `customers` → Confirm column `email`.
+                    4. Query:
+                       ```sql
+                       SELECT email FROM customers;
+
+                    ### Example 3 — Memory Usage:
+                    **User Request:** "Get top 5 most expensive products."
+
+                    **Steps:**
+                    1. Check memory for schema of `products` table.
+                    2. If column `price` exists in memory, directly generate:
+                       ```sql
+                       SELECT * FROM products ORDER BY price DESC LIMIT 5;
+
+                    ### Example 4 — COUNT Query:
+                    **User Request:** "How many orders have been placed?"
+
+                    **Steps:**
+                    1. List Tables → Identify a table like `orders`.
+                    2. Get Schema for `orders` → Confirm it's the right table.
+                    3. Query:
+                       ```sql
+                       SELECT COUNT(*) FROM orders;
+
+                    ### Example 5 — WHERE Clause (Filtering):
+                    **User Request:** "Get the names of customers from Germany."
+
+                    **Steps:**
+                    1. Use memory or List Tables → Identify `customers` table.
+                    2. Get Schema for `customers` → Confirm columns like `country`, `name`.
+                    3. Generate and execute query:
+                       ```sql
+                       SELECT name FROM customers WHERE country = 'Germany';
+
+                    ### Invalid Example — DELETE Operation (Not Allowed):
+                    **User Request:** "Delete all customers from Germany."
+
+                    **Response Guidance:**
+                    - **Do not generate or execute** destructive queries such as `DELETE`.
+                    - Instead, respond with a message like:
+                      > Destructive operations such as `DELETE` are not permitted. I can help you retrieve the customers from Germany using a `SELECT` query instead:
+                      > ```sql
+                      > SELECT * FROM customers WHERE country = 'Germany';
+                      > ```
+
+                    =========================
+                    # New User Request
+                    =========================
+
+                    Please fulfill the following request based on the above context:
+
+                    {new_request}
+                    """
+
+    return base_prompt
+
+
 @mcp.tool(description="tests the database connection and returns the PostgreSQL version or an error message.")
 async def test_connection(ctx: Context) -> str:
     """Test database connection"""

utils.py

@@ -0,0 +1,74 @@
+def parse_mcp_output(output_dict):
+    result = []
+    messages = output_dict.get("messages", [])
+
+    query_store = []
+
+    for msg in messages:
+        role_name = msg.__class__.__name__  # Example: HumanMessage, AIMessage, ToolMessage
+        content = getattr(msg, "content", "")
+
+        # AIMessage with tool call
+        if role_name == "AIMessage":
+            function_call = getattr(msg, "additional_kwargs", {}).get("function_call")
+            if function_call:
+                tool_name = function_call.get("name")
+                arguments = function_call.get("arguments")
+
+                # Check if arguments is a JSON string or a dict
+                if isinstance(arguments, str):
+                    import json
+                    try:
+                        arguments_dict = json.loads(arguments)
+                    except json.JSONDecodeError:
+                        arguments_dict = {}
+                else:
+                    arguments_dict = arguments or {}
+
+                # Check for presence of "query" key
+                if "query" in arguments_dict:
+                    print("query detected!!!")
+                    print(f"ai said:{content[0]}")
+                    print(arguments_dict["query"])
+                    query_store.append(arguments_dict["query"])
+
+                    result.append({
+                        "type": "ai_function_call",
+                        "ai_said": content,
+                        "tool": tool_name,
+                        "args": arguments
+                    })
+                else:
+                    print(f"ai said:{content}")
+                    result.append({
+                        "type": "ai_function_call",
+                        "ai_said": content,
+                        "tool": tool_name,
+                        "args": arguments
+                    })
+
+            else:
+                print(f"ai final answer:{content}")
+                result.append({
+                    "type": "ai_final_answer",
+                    "ai_said": content
+                })
+
+        # ToolMessage
+        elif role_name == "ToolMessage":
+            tool_name = getattr(msg, "name", None)
+            print(f"tool response:{content}")
+            result.append({
+                "type": "tool_response",
+                "tool": tool_name,
+                "response": content
+            })
+
+        elif role_name == "HumanMessage":
+            result.append({
+                "type": "user_message",
+                "content": content
+            })
+
+
+    return result, query_store