development-guides

Extend the Node base class, call super().__init__(), define parameters with NodeParameter for validation, and implement the execute() method returning a dictionary of outputs. For non-blocking operations, extend AsyncNode and implement execute_async() instead.

Use MCPServer from kailash.core.mcp_server, then register tools with the @server.tool decorator, resources with @server.resource, and prompts with @server.prompt. Run the server over stdio for Claude Desktop, HTTP for REST integrations, or WebSocket for real-time communication.

Use AsyncLocalRuntime for Docker and FastAPI deployments to avoid threading issues in async contexts. Use LocalRuntime for CLI scripts and synchronous contexts, or call get_runtime() for automatic detection.

Cyclic workflows are a planned feature not yet implemented in SDK v0.9.31. Current alternatives include Python loops inside nodes, recursive workflow execution, or SwitchNode-based state machine patterns.

Create one shared pool per worker at startup and pass it to AsyncSQLDatabaseNode via the external_pool parameter. Set max_pool_size to max database connections divided by worker count, and match the pool type to your database (asyncpg for PostgreSQL, aiomysql for MySQL).

The testing strategy requires real infrastructure in Tier 2 integration and Tier 3 end-to-end tests to catch actual behavior issues. Use in-memory SQLite databases and real test APIs like jsonplaceholder instead of mocks for reliable validation.