SDK Reference - Python

API reference for Flutch SDK in Python. Core classes and services for building graphs with FastAPI.

Installation

bash
actions.copy
flutch graph create my-graph --language python
cd my-graph
pip install -r requirements.txt

Main imports:

python
actions.copy
from flutch_sdk import AbstractGraphBuilder, callback, with_callbacks
from flutch_sdk import ModelInitializer, McpRuntimeClient

Graph Builder

Base class for all graph versions. Every graph version must inherit from AbstractGraphBuilder.

Basic Structure

python
actions.copy
from flutch_sdk import AbstractGraphBuilder, GraphRequestPayload

class MyGraphV1Builder(AbstractGraphBuilder):
    version = "1.0.0"

    def __init__(
        self,
        checkpointer,
        generate_node
    ):
        super().__init__()
        self.checkpointer = checkpointer
        self.generate_node = generate_node

    async def build_graph(self, payload=None):
        # Build your workflow here
        # See framework-specific guides for details
        return compiled_graph

Key methods:

Method	Description
build_graph(payload=None)	Build and return compiled workflow
prepare_config(payload)	Prepare config (auto-handled, override if needed)

Key properties:

Property	Type	Description
version	str	Semantic version (required)
graph_type	str	Auto-generated from manifest (e.g., "acme.support::1.0.0")
manifest	dict	Loaded graph manifest
logger	Logger	Python logger instance

Models

Initialize AI models in your graph nodes.

ModelInitializer

python
actions.copy
from flutch_sdk import ModelInitializer
from langchain_core.messages import SystemMessage
import logging

class GenerateNode:
    def __init__(self, model_initializer: ModelInitializer):
        self.model_initializer = model_initializer
        self.logger = logging.getLogger(__name__)

    async def execute(self, state: dict, config: dict = None) -> dict:
        # Get model settings from agent config
        graph_settings = config.get("configurable", {}).get("graph_settings", {})
        model_settings = graph_settings.get("model_settings", {})
        model_id = model_settings.get("model_id", "gpt-4o")
        temperature = model_settings.get("temperature", 0.7)

        self.logger.debug(f"Using model: {model_id}, temperature: {temperature}")

        # Initialize chat model
        model = await self.model_initializer.initialize_chat_model(
            model_id=model_id,
            temperature=temperature
        )

        # Use the model
        result = await model.ainvoke(state["messages"], config)

        return {"messages": [result]}

Methods:

Method	Parameters	Returns
initialize_chat_model(model_id, temperature?, max_tokens?)	Keyword args	BaseChatModel
initialize_embedding_model(model_id)	model_id: str	Embeddings
initialize_rerank_model(model_id, max_tokens?)	Keyword args	BaseDocumentCompressor

Example with tools:

python
actions.copy
# Initialize model
model = await self.model_initializer.initialize_chat_model(
    model_id="gpt-4o",
    temperature=0.7
)

# Bind tools to model
model_with_tools = model.bind_tools(tools)

# Invoke with tools
result = await model_with_tools.ainvoke(messages, config)

Tools

Execute MCP tools from your graph nodes.

McpRuntimeClient

python
actions.copy
from flutch_sdk import ModelInitializer, McpRuntimeClient
import logging

class GenerateNode:
    def __init__(
        self,
        model_initializer: ModelInitializer,
        mcp_client: McpRuntimeClient
    ):
        self.model_initializer = model_initializer
        self.mcp_client = mcp_client
        self.logger = logging.getLogger(__name__)

    async def execute(self, state: dict, config: dict = None) -> dict:
        graph_settings = config.get("configurable", {}).get("graph_settings", {})
        model_settings = graph_settings.get("model_settings", {})
        model_id = model_settings.get("model_id")
        enabled_tools = graph_settings.get("available_tools", [])

        # Initialize model
        model = await self.model_initializer.initialize_chat_model(
            model_id=model_id
        )

        # Get tools (filtered and converted automatically)
        if enabled_tools:
            tools = await self.mcp_client.get_tools(enabled_tools)
            model = model.bind_tools(tools)

            self.logger.debug(f"Configured {len(tools)} tools")

        # Invoke model with tools
        result = await model.ainvoke(state["messages"], config)

        return {"messages": [result]}

Methods:

Method	Parameters	Returns
get_tools(enabled_tools=None)	list[str] or None	list[StructuredTool]
execute_tool(name, args)	name: str, args: dict	McpToolResult

Execute tool calls:

python
actions.copy
async def execute_tools(self, state: dict, config: dict = None) -> dict:
    last_message = state["messages"][-1]
    tool_calls = getattr(last_message, "tool_calls", [])

    tool_messages = []
    for tool_call in tool_calls:
        result = await self.mcp_client.execute_tool(
            tool_call["name"],
            tool_call["args"]
        )

        tool_messages.append({
            "type": "tool",
            "tool_call_id": tool_call["id"],
            "content": result.get("data") if result["success"] else result.get("error"),
            "name": tool_call["name"]
        })

        self.logger.debug(f"Tool {tool_call['name']} executed successfully")

    return {"messages": tool_messages}

Callbacks

Add interactive buttons that execute your code when clicked.

Callback Decorators

1. Create callback handler class:

python
actions.copy
from flutch_sdk import callback, CallbackContext, CallbackResult
import logging

class MyCallbacks:
    def __init__(self):
        self.logger = logging.getLogger(__name__)

    @callback("confirm-payment")
    async def handle_confirm_payment(
        self,
        context: CallbackContext
    ) -> CallbackResult:
        amount = context.params.get("amount")
        order_id = context.params.get("order_id")

        self.logger.debug(f"Processing payment: {order_id}, amount: {amount}")

        try:
            await self.process_payment(order_id, amount)

            return {
                "success": True,
                "message": "Payment confirmed!",
                "patch": {
                    "text": "✅ Payment processed successfully",
                    "disable_buttons": True
                }
            }
        except Exception as error:
            self.logger.error(f"Payment failed: {str(error)}")

            return {
                "success": False,
                "error": "Payment processing failed. Please try again."
            }

    async def process_payment(self, order_id: str, amount: float):
        # Payment processing logic
        pass

2. Register callbacks in builder:

python
actions.copy
from flutch_sdk import AbstractGraphBuilder, with_callbacks
from .callbacks.my_callbacks import MyCallbacks

@with_callbacks(MyCallbacks)
class MyGraphV1Builder(AbstractGraphBuilder):
    version = "1.0.0"

    # Builder implementation...

Callback context:

python
actions.copy
class CallbackContext(TypedDict):
    user_id: str           # User who clicked
    thread_id: str         # Conversation thread
    agent_id: str          # Agent ID
    params: dict           # Parameters from callback token
    platform: str          # Platform (web, telegram, etc.)
    builder: Any           # Access to builder instance

Callback result:

python
actions.copy
class CallbackResult(TypedDict, total=False):
    success: bool
    message: str
    error: str
    patch: dict  # {"text": str, "disable_buttons": bool}

Configuration

Access agent configuration in your graph nodes.

Runtime Configuration

Agent configuration is injected through config["configurable"]["graph_settings"].

python
actions.copy
async def execute(self, state: dict, config: dict = None) -> dict:
    # Get all settings
    graph_settings = config.get("configurable", {}).get("graph_settings", {})

    # Access specific settings
    system_prompt = graph_settings.get("system_prompt", "")
    model_settings = graph_settings.get("model_settings", {})
    enabled_tools = graph_settings.get("available_tools", [])
    custom_settings = graph_settings.get("custom_settings", {})

    self.logger.debug(f"Model: {model_settings.get('model_id')}")
    self.logger.debug(f"Temperature: {model_settings.get('temperature')}")
    self.logger.debug(f"System prompt: {system_prompt}")
    self.logger.debug(f"Enabled tools: {enabled_tools}")

    # Use configuration
    model = await self.model_initializer.initialize_chat_model(
        model_id=model_settings.get("model_id"),
        temperature=model_settings.get("temperature")
    )

Settings structure (from config-schema.json):

python
actions.copy
class GraphSettings(TypedDict, total=False):
    system_prompt: str
    model_settings: ModelSettings
    available_tools: list[str]
    custom_settings: dict[str, Any]

class ModelSettings(TypedDict, total=False):
    model_id: str
    temperature: float
    max_tokens: int

Application Setup

Configure your graph application with FastAPI.

Main Application

python
actions.copy
# src/main.py
from flutch_sdk import bootstrap
from .graph_module import MyGraphModule

async def main():
    app = await bootstrap(MyGraphModule)
    return app

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "main:main",
        host="0.0.0.0",
        port=3000,
        reload=True
    )

Graph Module

python
actions.copy
# src/graph_module.py
from flutch_sdk import GraphModule
from .versions.v1_0_0.builder import MyGraphV1Builder
from .versions.v2_0_0.builder import MyGraphV2Builder
from .nodes.generate_node import GenerateNode
from .callbacks.my_callbacks import MyCallbacks

class MyGraphModule(GraphModule):
    def configure(self):
        self.register_builders([
            MyGraphV1Builder,
            MyGraphV2Builder
        ])

        self.register_providers([
            GenerateNode,
            MyCallbacks
        ])

The bootstrap() function:

• Initializes FastAPI application
• Registers all graph versions
• Starts HTTP server
• Sets up health checks
• Configures callback handlers