Update langgraph/mcp_output/mcp_plugin/mcp_service.py

langgraph/mcp_output/mcp_plugin/mcp_service.py

@@ -1,443 +1,192 @@
-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
 
-# 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
+# 创建 FastMCP 服务应用
 mcp = FastMCP("langgraph_service")
 
-# 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]:
+@mcp.tool(name="list_available_features", description="列出所有可用的 LangGraph 核心功能")
+def list_available_features() -> dict:
     """
-    Create a new StateGraph for building stateful workflows.
+    列出 LangGraph 的所有核心功能。
 
-    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
+    返回：
+    - dict: 包含 success 状态和功能列表。
     """
     try:
-        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
-        }
-        
+        features = [
+            "list_available_features",
+            "get_feature_info",
+            "execute_workflow",
+            "create_state_graph",
+            "validate_graph",
+            "draw_graph",
+            "deploy_agent"
+        ]
         return {
             "success": True,
-            "result": {
-                "graph_id": graph_id,
-                "state_schema": state_schema or {"messages": "List[str]"}
-            },
-            "error": None
+            "features": features,
+            "count": len(features)
         }
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
+        return {"success": False, "error": str(e)}
 
-@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]:
+@mcp.tool(name="get_feature_info", description="获取特定功能的详细信息")
+def get_feature_info(feature_name: str) -> dict:
     """
-    Add a node to a StateGraph.
+    获取 LangGraph 中某个功能的详细信息。
 
-    Args:
-        graph_id: ID of the graph
-        node_name: Name of the node
-        node_type: Type of node (function, tool, etc.)
+    参数：
+    - feature_name: 功能名称。
 
-    Returns:
-        Dictionary with success status
+    返回：
+    - dict: 包含功能的详细信息。
     """
     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"}
-        
-        # 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)
-        
+        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} 不存在。"}
         return {
             "success": True,
-            "result": {
-                "graph_id": graph_id,
-                "node_name": node_name,
-                "node_type": node_type
-            },
-            "error": None
+            "feature_name": feature_name,
+            "description": feature_info[feature_name]
         }
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
+        return {"success": False, "error": str(e)}
 
-@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]:
+@mcp.tool(name="execute_workflow", description="执行 LangGraph 的工作流")
+def execute_workflow(workflow_definition: dict) -> dict:
     """
-    Add an edge connecting two nodes in a graph.
+    执行 LangGraph 的工作流。
 
-    Args:
-        graph_id: ID of the graph
-        from_node: Source node name
-        to_node: Target node name (use "__end__" for terminal node)
+    参数：
+    - workflow_definition: 工作流的定义，包括节点和边的信息。
 
-    Returns:
-        Dictionary with success status
+    返回：
+    - dict: 包含执行结果或错误信息。
     """
     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"}
-        
-        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)}
+        from langgraph.graph import StateGraph
 
+        # 创建状态图
+        graph = StateGraph(workflow_definition.get("state", {}))
 
-@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.
+        # 添加节点
+        for node_name, node_func in workflow_definition.get("nodes", {}).items():
+            graph.add_node(node_name, node_func)
 
-    Args:
-        graph_id: ID of the graph
-        from_node: Source node name
-        condition_type: Type of condition (simple, multi, etc.)
+        # 添加边
+        for edge in workflow_definition.get("edges", []):
+            graph.add_edge(edge[0], edge[1])
 
-    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, "result": None, "error": str(e)}
+        # 编译并执行工作流
+        compiled_graph = graph.compile()
+        result = compiled_graph.invoke(workflow_definition.get("input", {}))
 
+        return {"success": True, "result": result}
+    except Exception as e:
+        return {"success": False, "error": str(e)}
 
-@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]:
+@mcp.tool(name="create_state_graph", description="创建一个状态图")
+def create_state_graph(graph_definition: dict) -> dict:
     """
-    Set the entry point (starting node) for a graph.
+    创建一个状态图。
 
-    Args:
-        graph_id: ID of the graph
-        node_name: Name of the entry node
+    参数：
+    - graph_definition: 图的定义，包括节点和边的信息。
 
-    Returns:
-        Dictionary with success status
+    返回：
+    - dict: 包含状态图的定义或错误信息。
     """
     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"}
-        
-        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)}
+        from langgraph.graph import StateGraph
 
+        # 创建状态图
+        graph = StateGraph(graph_definition.get("state", {}))
 
-@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 node_name, node_func in graph_definition.get("nodes", {}).items():
+            graph.add_node(node_name, node_func)
 
-    Args:
-        graph_id: ID of the graph to compile
+        # 添加边
+        for edge in graph_definition.get("edges", []):
+            graph.add_edge(edge[0], edge[1])
 
-    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
-        }
+        return {"success": True, "graph": graph}
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
+        return {"success": False, "error": str(e)}
 
-@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]:
+@mcp.tool(name="validate_graph", description="验证图的正确性")
+def validate_graph(graph: object) -> dict:
     """
-    Execute a compiled graph with given input.
+    验证图的正确性。
 
-    Args:
-        graph_id: ID of the compiled graph
-        input_data: Input state/data for the graph
+    参数：
+    - graph: 要验证的图对象。
 
-    Returns:
-        Dictionary with execution result
+    返回：
+    - dict: 包含验证结果或错误信息。
     """
     try:
-        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, "result": None, "error": str(e)}
+        from langgraph.pregel._validate import validate_graph
 
+        # 验证图
+        validate_graph(graph)
+        return {"success": True, "message": "图验证通过"}
+    except Exception as e:
+        return {"success": False, "error": str(e)}
 
-@mcp.tool(name="get_graph_info", description="Get information about a graph")
-def get_graph_info(graph_id: str) -> Dict[str, Any]:
+@mcp.tool(name="draw_graph", description="可视化图结构")
+def draw_graph(graph: object) -> dict:
     """
-    Get detailed information about a graph.
+    可视化图结构。
 
-    Args:
-        graph_id: ID of the graph
+    参数：
+    - graph: 要可视化的图对象。
 
-    Returns:
-        Dictionary with graph details
+    返回：
+    - dict: 包含可视化结果或错误信息。
     """
     try:
-        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)}
-
+        from langgraph.pregel._draw import draw_graph
 
-@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
-        }
+        # 绘制图
+        draw_graph(graph)
+        return {"success": True, "message": "图已绘制"}
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        return {"success": False, "error": str(e)}
 
-
-@mcp.tool(name="delete_graph", description="Delete a graph")
-def delete_graph(graph_id: str) -> Dict[str, Any]:
+@mcp.tool(name="deploy_agent", description="部署 LangGraph 的代理")
+def deploy_agent(agent_config: dict) -> dict:
     """
-    Delete a stored graph.
+    部署 LangGraph 的代理。
 
-    Args:
-        graph_id: ID of the graph to delete
+    参数：
+    - agent_config: 代理的配置，包括名称和参数。
 
-    Returns:
-        Dictionary with success status
+    返回：
+    - dict: 包含部署结果或错误信息。
     """
     try:
-        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, "result": None, "error": str(e)}
+        # 示例：假设有一个部署代理的函数
+        from langgraph.agent import deploy_agent
 
+        deploy_agent(agent_config)
+        return {"success": True, "message": "代理已部署"}
+    except Exception as e:
+        return {"success": False, "error": str(e)}
 
 def create_app() -> FastMCP:
     """
-    Create and return the FastMCP application instance.
+    创建并返回 FastMCP 应用实例。
 
-    Returns:
-        FastMCP: The FastMCP application instance.
+    返回：
+    - FastMCP: FastMCP 应用实例。
     """
     return mcp