diff --git a/src/backend/base/langflow/graph/edge/schema.py b/src/backend/base/langflow/graph/edge/schema.py
index 628073d1326..d8ae9963c18 100644
--- a/src/backend/base/langflow/graph/edge/schema.py
+++ b/src/backend/base/langflow/graph/edge/schema.py
@@ -1,5 +1,9 @@
-from typing import Optional, Any, List
-from pydantic import BaseModel
+from typing import Any, List, Optional
+
+from pydantic import Field, field_validator
+from typing_extensions import TypedDict
+
+from langflow.helpers.base_model import BaseModel
 
 
 class ResultPair(BaseModel):
@@ -32,3 +36,59 @@ def format(self, sep: str = "\n") -> str:
                 for result_pair in self.result_pairs[:-1]
             ]
         )
+
+
+class TargetHandle(BaseModel):
+    fieldName: str = Field(..., alias="fieldName", description="Field name for the target handle.")
+    id: str = Field(..., description="Unique identifier for the target handle.")
+    input_types: List[str] = Field(
+        default_factory=list, alias="inputTypes", description="List of input types for the target handle."
+    )
+    type: str = Field(..., description="Type of the target handle.")
+
+
+class SourceHandle(BaseModel):
+    base_classes: list[str] = Field(
+        default_factory=list, alias="baseClasses", description="List of base classes for the source handle."
+    )
+    data_type: str = Field(..., alias="dataType", description="Data type for the source handle.")
+    id: str = Field(..., description="Unique identifier for the source handle.")
+    name: Optional[str] = Field(None, description="Name of the source handle.")
+    output_types: List[str] = Field(default_factory=list, description="List of output types for the source handle.")
+
+    @field_validator("name", mode="before")
+    @classmethod
+    def validate_name(cls, v, _info):
+        if _info.data["data_type"] == "GroupNode":
+            # 'OpenAIModel-u4iGV_text_output'
+            splits = v.split("_", 1)
+            if len(splits) != 2:
+                raise ValueError(f"Invalid source handle name {v}")
+            v = splits[1]
+        return v
+
+
+class SourceHandleDict(TypedDict, total=False):
+    baseClasses: list[str]
+    dataType: str
+    id: str
+    name: Optional[str]
+    output_types: List[str]
+
+
+class TargetHandleDict(TypedDict):
+    fieldName: str
+    id: str
+    inputTypes: Optional[List[str]]
+    type: str
+
+
+class EdgeDataDetails(TypedDict):
+    sourceHandle: SourceHandleDict
+    targetHandle: TargetHandleDict
+
+
+class EdgeData(TypedDict, total=False):
+    source: str
+    target: str
+    data: EdgeDataDetails
diff --git a/src/backend/base/langflow/helpers/base_model.py b/src/backend/base/langflow/helpers/base_model.py
new file mode 100644
index 00000000000..c81fd99d2c6
--- /dev/null
+++ b/src/backend/base/langflow/helpers/base_model.py
@@ -0,0 +1,6 @@
+from pydantic import BaseModel as PydanticBaseModel
+from pydantic import ConfigDict
+
+
+class BaseModel(PydanticBaseModel):
+    model_config = ConfigDict(populate_by_name=True)