Update langgraph/mcp_output/mcp_plugin/mcp_service.py

langgraph/mcp_output/mcp_plugin/mcp_service.py

@@ -1,192 +1,443 @@
+import os
+import sys
+from typing import Dict, Any, List, Optional, Union
+import json
+
+# Add the local source directory to sys.path
+source_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "source")
+if source_path not in sys.path:
+    sys.path.insert(0, source_path)
+
 from fastmcp import FastMCP
 
-# 创建 FastMCP 服务应用
+# Import LangGraph modules
+try:
+    from langgraph.graph import StateGraph, Graph, END
+    from langgraph.prebuilt import ToolExecutor, ToolInvocation
+    from typing_extensions import TypedDict
+except ImportError:
+    # Fallback for basic functionality
+    StateGraph = None
+    Graph = None
+    END = "__end__"
+
+# Create the FastMCP service application
 mcp = FastMCP("langgraph_service")
 
-@mcp.tool(name="list_available_features", description="列出所有可用的 LangGraph 核心功能")
-def list_available_features() -> dict:
+# Storage for graphs
+_graphs: Dict[str, Any] = {}
+_compiled_graphs: Dict[str, Any] = {}
+
+
+@mcp.tool(name="create_state_graph", description="Create a new StateGraph")
+def create_state_graph(
+    graph_id: str,
+    state_schema: Optional[Dict[str, str]] = None
+) -> Dict[str, Any]:
     """
-    列出 LangGraph 的所有核心功能。
+    Create a new StateGraph for building stateful workflows.
 
-    返回：
-    - dict: 包含 success 状态和功能列表。
+    Args:
+        graph_id: Unique identifier for the graph
+        state_schema: Optional schema defining state structure (dict of field_name: type_name)
+
+    Returns:
+        Dictionary with success status and graph info
     """
     try:
-        features = [
-            "list_available_features",
-            "get_feature_info",
-            "execute_workflow",
-            "create_state_graph",
-            "validate_graph",
-            "draw_graph",
-            "deploy_agent"
-        ]
+        if StateGraph is None:
+            return {"success": False, "result": None, "error": "StateGraph not available"}
+        
+        # Create a simple state type
+        if state_schema:
+            # Create a TypedDict-like state class dynamically
+            state_fields = {k: str for k, v in state_schema.items()}
+            
+            class GraphState(TypedDict):
+                pass
+            
+            # Add fields
+            for field, ftype in state_fields.items():
+                GraphState.__annotations__[field] = Any
+            
+            graph = StateGraph(GraphState)
+        else:
+            # Default state with messages
+            class DefaultState(TypedDict):
+                messages: List[str]
+            
+            graph = StateGraph(DefaultState)
+        
+        _graphs[graph_id] = {
+            "graph": graph,
+            "nodes": [],
+            "edges": [],
+            "compiled": False
+        }
+        
         return {
             "success": True,
-            "features": features,
-            "count": len(features)
+            "result": {
+                "graph_id": graph_id,
+                "state_schema": state_schema or {"messages": "List[str]"}
+            },
+            "error": None
         }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
 
-@mcp.tool(name="get_feature_info", description="获取特定功能的详细信息")
-def get_feature_info(feature_name: str) -> dict:
+@mcp.tool(name="add_node", description="Add a node to a graph")
+def add_node(
+    graph_id: str,
+    node_name: str,
+    node_type: str = "function"
+) -> Dict[str, Any]:
     """
-    获取 LangGraph 中某个功能的详细信息。
+    Add a node to a StateGraph.
 
-    参数：
-    - feature_name: 功能名称。
+    Args:
+        graph_id: ID of the graph
+        node_name: Name of the node
+        node_type: Type of node (function, tool, etc.)
 
-    返回：
-    - dict: 包含功能的详细信息。
+    Returns:
+        Dictionary with success status
     """
     try:
-        feature_info = {
-            "list_available_features": "列出所有可用的 LangGraph 核心功能。",
-            "get_feature_info": "获取特定功能的详细信息。",
-            "execute_workflow": "执行 LangGraph 的工作流。",
-            "create_state_graph": "创建一个状态图。",
-            "validate_graph": "验证图的正确性。",
-            "draw_graph": "可视化图结构。",
-            "deploy_agent": "部署 LangGraph 的代理。"
-        }
-        if feature_name not in feature_info:
-            return {"success": False, "error": f"功能 {feature_name} 不存在。"}
+        if graph_id not in _graphs:
+            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
+        
+        graph_data = _graphs[graph_id]
+        
+        if graph_data["compiled"]:
+            return {"success": False, "result": None, "error": "Cannot modify compiled graph"}
+        
+        # Create a simple node function
+        def node_function(state):
+            # Default: pass through state with a message
+            if "messages" in state:
+                messages = state.get("messages", [])
+                messages.append(f"Processed by {node_name}")
+                return {"messages": messages}
+            return state
+        
+        graph_data["graph"].add_node(node_name, node_function)
+        graph_data["nodes"].append(node_name)
+        
         return {
             "success": True,
-            "feature_name": feature_name,
-            "description": feature_info[feature_name]
+            "result": {
+                "graph_id": graph_id,
+                "node_name": node_name,
+                "node_type": node_type
+            },
+            "error": None
         }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
 
-@mcp.tool(name="execute_workflow", description="执行 LangGraph 的工作流")
-def execute_workflow(workflow_definition: dict) -> dict:
+@mcp.tool(name="add_edge", description="Add an edge between two nodes")
+def add_edge(
+    graph_id: str,
+    from_node: str,
+    to_node: str
+) -> Dict[str, Any]:
     """
-    执行 LangGraph 的工作流。
+    Add an edge connecting two nodes in a graph.
 
-    参数：
-    - workflow_definition: 工作流的定义，包括节点和边的信息。
+    Args:
+        graph_id: ID of the graph
+        from_node: Source node name
+        to_node: Target node name (use "__end__" for terminal node)
 
-    返回：
-    - dict: 包含执行结果或错误信息。
+    Returns:
+        Dictionary with success status
     """
     try:
-        from langgraph.graph import StateGraph
-
-        # 创建状态图
-        graph = StateGraph(workflow_definition.get("state", {}))
+        if graph_id not in _graphs:
+            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
+        
+        graph_data = _graphs[graph_id]
+        
+        if graph_data["compiled"]:
+            return {"success": False, "result": None, "error": "Cannot modify compiled graph"}
+        
+        graph_data["graph"].add_edge(from_node, to_node)
+        graph_data["edges"].append({"from": from_node, "to": to_node})
+        
+        return {
+            "success": True,
+            "result": {
+                "graph_id": graph_id,
+                "from_node": from_node,
+                "to_node": to_node
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
 
-        # 添加节点
-        for node_name, node_func in workflow_definition.get("nodes", {}).items():
-            graph.add_node(node_name, node_func)
 
-        # 添加边
-        for edge in workflow_definition.get("edges", []):
-            graph.add_edge(edge[0], edge[1])
+@mcp.tool(name="add_conditional_edge", description="Add a conditional edge with routing logic")
+def add_conditional_edge(
+    graph_id: str,
+    from_node: str,
+    condition_type: str = "simple"
+) -> Dict[str, Any]:
+    """
+    Add a conditional edge that routes based on state.
 
-        # 编译并执行工作流
-        compiled_graph = graph.compile()
-        result = compiled_graph.invoke(workflow_definition.get("input", {}))
+    Args:
+        graph_id: ID of the graph
+        from_node: Source node name
+        condition_type: Type of condition (simple, multi, etc.)
 
-        return {"success": True, "result": result}
+    Returns:
+        Dictionary with success status
+    """
+    try:
+        if graph_id not in _graphs:
+            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
+        
+        graph_data = _graphs[graph_id]
+        
+        if graph_data["compiled"]:
+            return {"success": False, "result": None, "error": "Cannot modify compiled graph"}
+        
+        # Simple routing function
+        def route_function(state):
+            # Default routing based on message count
+            messages = state.get("messages", [])
+            if len(messages) > 3:
+                return END
+            else:
+                # Route to first available node or END
+                nodes = graph_data["nodes"]
+                return nodes[0] if nodes else END
+        
+        graph_data["graph"].add_conditional_edges(
+            from_node,
+            route_function
+        )
+        
+        return {
+            "success": True,
+            "result": {
+                "graph_id": graph_id,
+                "from_node": from_node,
+                "condition_type": condition_type
+            },
+            "error": None
+        }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
 
-@mcp.tool(name="create_state_graph", description="创建一个状态图")
-def create_state_graph(graph_definition: dict) -> dict:
+@mcp.tool(name="set_entry_point", description="Set the entry point of the graph")
+def set_entry_point(
+    graph_id: str,
+    node_name: str
+) -> Dict[str, Any]:
     """
-    创建一个状态图。
+    Set the entry point (starting node) for a graph.
 
-    参数：
-    - graph_definition: 图的定义，包括节点和边的信息。
+    Args:
+        graph_id: ID of the graph
+        node_name: Name of the entry node
 
-    返回：
-    - dict: 包含状态图的定义或错误信息。
+    Returns:
+        Dictionary with success status
     """
     try:
-        from langgraph.graph import StateGraph
+        if graph_id not in _graphs:
+            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
+        
+        graph_data = _graphs[graph_id]
+        
+        if graph_data["compiled"]:
+            return {"success": False, "result": None, "error": "Cannot modify compiled graph"}
+        
+        graph_data["graph"].set_entry_point(node_name)
+        
+        return {
+            "success": True,
+            "result": {
+                "graph_id": graph_id,
+                "entry_point": node_name
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
 
-        # 创建状态图
-        graph = StateGraph(graph_definition.get("state", {}))
 
-        # 添加节点
-        for node_name, node_func in graph_definition.get("nodes", {}).items():
-            graph.add_node(node_name, node_func)
+@mcp.tool(name="compile_graph", description="Compile a graph for execution")
+def compile_graph(
+    graph_id: str
+) -> Dict[str, Any]:
+    """
+    Compile a StateGraph into an executable workflow.
 
-        # 添加边
-        for edge in graph_definition.get("edges", []):
-            graph.add_edge(edge[0], edge[1])
+    Args:
+        graph_id: ID of the graph to compile
 
-        return {"success": True, "graph": graph}
+    Returns:
+        Dictionary with success status and compiled graph info
+    """
+    try:
+        if graph_id not in _graphs:
+            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
+        
+        graph_data = _graphs[graph_id]
+        
+        if graph_data["compiled"]:
+            return {"success": False, "result": None, "error": "Graph already compiled"}
+        
+        compiled = graph_data["graph"].compile()
+        _compiled_graphs[graph_id] = compiled
+        graph_data["compiled"] = True
+        
+        return {
+            "success": True,
+            "result": {
+                "graph_id": graph_id,
+                "num_nodes": len(graph_data["nodes"]),
+                "num_edges": len(graph_data["edges"]),
+                "compiled": True
+            },
+            "error": None
+        }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
 
-@mcp.tool(name="validate_graph", description="验证图的正确性")
-def validate_graph(graph: object) -> dict:
+@mcp.tool(name="invoke_graph", description="Execute a compiled graph with input")
+def invoke_graph(
+    graph_id: str,
+    input_data: Dict[str, Any]
+) -> Dict[str, Any]:
     """
-    验证图的正确性。
+    Execute a compiled graph with given input.
 
-    参数：
-    - graph: 要验证的图对象。
+    Args:
+        graph_id: ID of the compiled graph
+        input_data: Input state/data for the graph
 
-    返回：
-    - dict: 包含验证结果或错误信息。
+    Returns:
+        Dictionary with execution result
     """
     try:
-        from langgraph.pregel._validate import validate_graph
-
-        # 验证图
-        validate_graph(graph)
-        return {"success": True, "message": "图验证通过"}
+        if graph_id not in _compiled_graphs:
+            return {"success": False, "result": None, "error": f"Compiled graph '{graph_id}' not found"}
+        
+        compiled_graph = _compiled_graphs[graph_id]
+        result = compiled_graph.invoke(input_data)
+        
+        return {
+            "success": True,
+            "result": {
+                "graph_id": graph_id,
+                "output": result
+            },
+            "error": None
+        }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
 
-@mcp.tool(name="draw_graph", description="可视化图结构")
-def draw_graph(graph: object) -> dict:
+@mcp.tool(name="get_graph_info", description="Get information about a graph")
+def get_graph_info(graph_id: str) -> Dict[str, Any]:
     """
-    可视化图结构。
+    Get detailed information about a graph.
 
-    参数：
-    - graph: 要可视化的图对象。
+    Args:
+        graph_id: ID of the graph
 
-    返回：
-    - dict: 包含可视化结果或错误信息。
+    Returns:
+        Dictionary with graph details
     """
     try:
-        from langgraph.pregel._draw import draw_graph
+        if graph_id not in _graphs:
+            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
+        
+        graph_data = _graphs[graph_id]
+        
+        return {
+            "success": True,
+            "result": {
+                "graph_id": graph_id,
+                "nodes": graph_data["nodes"],
+                "edges": graph_data["edges"],
+                "compiled": graph_data["compiled"]
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
 
-        # 绘制图
-        draw_graph(graph)
-        return {"success": True, "message": "图已绘制"}
+@mcp.tool(name="list_graphs", description="List all stored graphs")
+def list_graphs() -> Dict[str, Any]:
+    """
+    List all stored graphs.
+
+    Returns:
+        Dictionary with list of graph IDs
+    """
+    try:
+        graphs_info = []
+        for gid, gdata in _graphs.items():
+            graphs_info.append({
+                "graph_id": gid,
+                "num_nodes": len(gdata["nodes"]),
+                "num_edges": len(gdata["edges"]),
+                "compiled": gdata["compiled"]
+            })
+        
+        return {
+            "success": True,
+            "result": {"graphs": graphs_info},
+            "error": None
+        }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="deploy_agent", description="部署 LangGraph 的代理")
-def deploy_agent(agent_config: dict) -> dict:
+
+@mcp.tool(name="delete_graph", description="Delete a graph")
+def delete_graph(graph_id: str) -> Dict[str, Any]:
     """
-    部署 LangGraph 的代理。
+    Delete a stored graph.
 
-    参数：
-    - agent_config: 代理的配置，包括名称和参数。
+    Args:
+        graph_id: ID of the graph to delete
 
-    返回：
-    - dict: 包含部署结果或错误信息。
+    Returns:
+        Dictionary with success status
     """
     try:
-        # 示例：假设有一个部署代理的函数
-        from langgraph.agent import deploy_agent
-
-        deploy_agent(agent_config)
-        return {"success": True, "message": "代理已部署"}
+        if graph_id not in _graphs:
+            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
+        
+        del _graphs[graph_id]
+        if graph_id in _compiled_graphs:
+            del _compiled_graphs[graph_id]
+        
+        return {
+            "success": True,
+            "result": {"deleted": graph_id},
+            "error": None
+        }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
 
 def create_app() -> FastMCP:
     """
-    创建并返回 FastMCP 应用实例。
+    Create and return the FastMCP application instance.
 
-    返回：
-    - FastMCP: FastMCP 应用实例。
+    Returns:
+        FastMCP: The FastMCP application instance.
     """
     return mcp