Feature/openapi automation

examples/openapi_automation/models.py

@@ -0,0 +1,111 @@
+from typing import Any, Dict, List, Optional
+
+from pipelex.core.stuffs.structured_content import StructuredContent
+from pydantic import BaseModel, Field
+
+
+class FunctionParameter(StructuredContent):
+    name: str = Field(description="parameter name")
+    value: str = Field(description="parameter value")
+    type: str = Field(description="parameter type")
+
+
+# OpenAPI Specification Models
+class OpenAPIParameter(BaseModel):
+    name: str
+    in_: str = Field(alias="in")
+    required: Optional[bool] = False
+    description: Optional[str] = None
+    schema_: Optional[Dict[str, Any]] = Field(default=None, alias="schema")
+
+
+class OpenAPIRequestBody(BaseModel):
+    description: Optional[str] = None
+    required: Optional[bool] = False
+    content: Optional[Dict[str, Any]] = None
+
+
+class OpenAPIResponse(BaseModel):
+    description: Optional[str] = None
+    content: Optional[Dict[str, Any]] = None
+
+
+class OpenAPIOperation(BaseModel):
+    operationId: Optional[str] = None
+    summary: Optional[str] = None
+    description: Optional[str] = None
+    parameters: Optional[List[OpenAPIParameter]] = None
+    requestBody: Optional[OpenAPIRequestBody] = None
+    responses: Optional[Dict[str, OpenAPIResponse]] = None
+    tags: Optional[List[str]] = None
+
+
+class OpenAPIPathItem(BaseModel):
+    get: Optional[OpenAPIOperation] = None
+    post: Optional[OpenAPIOperation] = None
+    put: Optional[OpenAPIOperation] = None
+    delete: Optional[OpenAPIOperation] = None
+    patch: Optional[OpenAPIOperation] = None
+    options: Optional[OpenAPIOperation] = None
+    head: Optional[OpenAPIOperation] = None
+
+
+class OpenAPIInfo(BaseModel):
+    title: str
+    version: str
+    description: Optional[str] = None
+
+
+class OpenAPISpec(StructuredContent):
+    openapi: str = Field(description="OpenAPI version")
+    info: OpenAPIInfo = Field(description="API metadata")
+    paths: Dict[str, OpenAPIPathItem] = Field(description="API endpoints")
+    components: Optional[Dict[str, Any]] = Field(default=None, description="Reusable components")
+    servers: Optional[List[Dict[str, Any]]] = Field(default=None, description="API servers")
+
+
+class FunctionInfo(StructuredContent):
+    function_name: str = Field(description="The operation ID / function name")
+    description: Optional[str] = Field(default=None, description="Function description")
+
+
+class FunctionChoice(StructuredContent):
+    explanation: str = Field(description="Explanation of the choice.")
+    function_name: str = Field(description="Name of the function.")
+
+
+class ParameterDetail(StructuredContent):
+    """Detailed parameter information for API calls"""
+
+    name: str = Field(description="Parameter name")
+    param_in: str = Field(description="Where the parameter goes: path, query, header, cookie")
+    required: bool = Field(default=False, description="Whether the parameter is required")
+    param_type: Optional[str] = Field(default=None, description="Parameter data type")
+    description: Optional[str] = Field(default=None, description="Parameter description")
+    default: Optional[Any] = Field(default=None, description="Default value if any")
+
+
+class FunctionDetails(StructuredContent):
+    """Complete details needed to make an API request"""
+
+    function_name: str = Field(description="The operation ID / function name")
+    http_method: str = Field(description="HTTP method (GET, POST, PUT, DELETE, etc.)")
+    path: str = Field(description="API endpoint path")
+    description: Optional[str] = Field(default=None, description="Operation description")
+    parameters: List[ParameterDetail] = Field(default_factory=list, description="List of parameters")  # type: ignore[reportUnknownVariableType]
+    request_body_required: bool = Field(default=False, description="Whether a request body is required")
+    request_body_schema: Optional[Dict[str, Any]] = Field(default=None, description="Request body schema if applicable")
+    tags: Optional[List[str]] = Field(default=None, description="Operation tags")
+
+
+class RequestDetails(StructuredContent):
+    """Holds the actual parameter values needed to make an API request"""
+
+    function_name: str = Field(description="The operation ID / function name")
+    http_method: str = Field(description="HTTP method (GET, POST, PUT, DELETE, etc.)")
+    path: str = Field(description="API endpoint path")
+    query_parameters: Optional[Dict[str, Any]] = Field(default=None, description="Query parameters and their values")
+    path_parameters: Optional[Dict[str, Any]] = Field(default=None, description="Path parameters and their values")
+    header_parameters: Optional[Dict[str, Any]] = Field(default=None, description="Header parameters and their values")
+    cookie_parameters: Optional[Dict[str, Any]] = Field(default=None, description="Cookie parameters and their values")
+    request_body: Optional[Dict[str, Any]] = Field(default=None, description="Request body data if applicable")

examples/openapi_automation/openapi_automation.plx

@@ -0,0 +1,128 @@
+domain = "openapi_automation"
+description = "Building function info from OpenAPI JSON spec"
+main_pipe = "build_function_info"
+
+[concept]
+OpenAPISpec = "Structured OpenAPI specification for the backend."
+FunctionInfo = "Information about a function in the OpenAPI spec."
+FunctionChoice = "Choice of OpenAPI function to accomplish an operation."
+RequestDetails = "Request Details containing actual values for the request"
+
+[concept.OpenAPIURL]
+description = "The URL of the OpenAPI JSON spec"
+refines = "Text"
+
+[concept.OperationToAccomplish]
+description = "The specific operation to accomplish."
+refines = "Text"
+
+[concept.RelevantOpenapiPaths]
+description = """
+Relevant information (e.g., paths, methods) from the OpenAPI JSON specification that pertains to the operation to accomplish.
+"""
+
+[concept.RelevantOpenapiPaths.structure]
+paths = { type = "text", description = "List of relevant paths.", required = true }
+methods = { type = "text", description = "List of relevant methods.", required = true }
+
+[concept.FunctionName]
+description = "The name of the function."
+refines = "Text"
+
+[concept.FunctionParameter]
+description = "The necessary function parameters."
+
+[concept.FunctionParameter.structure]
+name = { type = "text", description = "Name of a function parameter.", required = true }
+type = { type = "text", description = "Data type of a function parameter.", required = true }
+value = { type = "text", description = "Values of each function parameter.", required = true }
+
+[concept.ApiResponseResult]
+description = "The compiled api response content."
+
+[concept.ApiResponseResult.structure]
+response = { type = "text", description = "The response of the api server", required = true }
+
+
+[pipe.build_function_info]
+type = "PipeSequence"
+description = """
+Main pipeline that builds the function name and function parameters and values necessary for the task based on the OpenAPI JSON spec and the operation to accomplish.
+"""
+inputs = { openapi_url = "OpenAPIURL", operation_to_accomplish = "OperationToAccomplish" }
+output = "ApiResponseResult"
+#output = "FunctionDetails"
+steps = [
+    { pipe = "obtain_api_spec", result = "openapi_spec"},
+    { pipe = "extract_available_functions", result = "function_info" },
+    { pipe = "choose_function", result = "function_choice" },
+    { pipe = "get_function_details", result = "function_details" },
+    { pipe = "prepare_request", result = "request_details"},
+    { pipe = "execute_api_call", result = "result_api_call" },
+]
+
+[pipe.obtain_api_spec]
+type = "PipeFunc"
+description = "Obtains the OpenAPI spec given a URL."
+inputs = { openapi_url = "OpenAPIURL" }
+output = "OpenAPISpec"
+function_name = "obtain_openapi_model"
+
+[pipe.extract_available_functions]
+type = "PipeFunc"
+description = "Extracts the available functions from the OpenAPI spec."
+inputs = { openapi_url = "OpenAPIURL" }
+output = "FunctionInfo[]"
+function_name = "extract_available_functions"
+
+
+[pipe.choose_function]
+type = "PipeLLM"
+description = "Uses the operation to accomplish and relevant OpenAPI paths to determine the function name."
+inputs = { operation_to_accomplish = "OperationToAccomplish", function_info = "FunctionInfo[]" }
+output = "FunctionChoice"
+model = "llm_to_engineer"
+system_prompt = """
+Determine a function name based on the operation to accomplish and relevant OpenAPI paths. Be concise.
+"""
+prompt = """
+Based on the operation to accomplish and the available OpenAPI functions, choose the relevant function name.
+
+@operation_to_accomplish
+
+@function_info
+"""
+
+[pipe.get_function_details]
+type = "PipeFunc"
+description = "Gets the details of a function from the OpenAPI spec."
+inputs = { function_choice = "FunctionChoice" }
+output = "FunctionDetails"
+function_name = "get_function_details"
+
+
+
+[pipe.prepare_request]
+type = "PipeLLM"
+description = "Prepares the request body corresponding to the actual request"
+inputs = {  operation_to_accomplish = "OperationToAccomplish", function_details = "FunctionDetails" }
+output = "RequestDetails"
+model = "llm_to_engineer"
+prompt = """
+Based on the operation to accomplish and the available OpenAPI functions, fill in the actual values for the current request.
+
+@operation_to_accomplish
+
+@function_details
+
+"""
+
+
+[pipe.execute_api_call]
+type = "PipeFunc"
+description = "Execute the API request given a CompiledFunctionInfo."
+inputs = { request_details = "RequestDetails" }
+output = "ApiResponseResult"
+function_name = "invoke_function_api_backend"
+
+

examples/openapi_automation/pipe_runner.py

@@ -0,0 +1,41 @@
+import asyncio
+
+from pipelex.pipelex import Pipelex
+from pipelex.pipeline.execute import execute_pipeline
+
+spec_url_1 = """
+https://developer.keap.com/docs/rest/2025-11-05-v1.json
+"""
+
+
+spec_url_2 = """
+https://petstore3.swagger.io/api/v3/openapi.json
+"""
+
+USE_CASE_1 = (spec_url_1, "extract all the contacts with email [email protected]")
+USE_CASE_2 = (spec_url_2, "get me the user with username: johndoe")
+
+
+async def run_build_function_info(use_case: tuple[str, str]):
+    spec_url, operation_to_accomplish = use_case
+    return await execute_pipeline(
+        pipe_code="build_function_info",
+        inputs={
+            "openapi_url": {
+                "concept": "openapi_function_builder.OpenAPIURL",
+                "content": spec_url,
+            },
+            "operation_to_accomplish": {
+                "concept": "openapi_function_builder.OperationToAccomplish",
+                "content": operation_to_accomplish,
+            },
+        },
+    )
+
+
+if __name__ == "__main__":
+    # Initialize Pipelex
+    Pipelex.make()
+
+    # Run the pipeline
+    result = asyncio.run(run_build_function_info(use_case=USE_CASE_2))