API Reference

from nadoo_flow import ChainableNode, NodeResult

class MyNode(ChainableNode):
    async def execute(self, node_context, workflow_context):
        return NodeResult(success=True, output={"result": "data"})

# Chain nodes with pipe operator
workflow = MyNode() | ProcessNode() | OutputNode()
result = await workflow.run({"input": "data"})

from nadoo_plugin import NadooPlugin, tool, parameter

class MyPlugin(NadooPlugin):
    @tool(name="process_text", description="Process text using AI")
    @parameter("text", type="string", required=True)
    def process_text(self, text: str) -> dict:
        # Use Nadoo's LLM
        response = self.api.llm.invoke(
            messages=[
                {"role": "user", "content": f"Process: {text}"}
            ]
        )

        return {"processed": response.content}

curl -X POST http://localhost:8002/api/v1/execute \
  -H "X-API-Key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "code": "print(\"Hello, World!\")",
    "language": "python"
  }'

# Flow Core
from nadoo_flow import ChainableNode, WorkflowExecutor

# Plugin SDK
from nadoo_plugin import NadooPlugin, tool, parameter

# For Sandbox, use REST API

# Async execution
result = await node.execute(node_context, workflow_context)
result = await workflow.run(input_data)

# Streaming
async for chunk in workflow.stream(input_data):
    process(chunk)

from typing import Dict, Any
from pydantic import BaseModel

class Config(BaseModel):
    timeout: float = 30.0
    retries: int = 3

from nadoo_flow import NodeResult

try:
    result = await node.execute(context)
    return NodeResult(success=True, output=data)
except Exception as e:
    return NodeResult(success=False, error=str(e))

curl -H "X-API-Key: your-api-key" ...