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 it with stdio transport for Claude Desktop, or HTTP/WebSocket for web integrations.

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.

Follow the 3-tier strategy: unit tests for individual nodes, integration tests against real infrastructure like in-memory SQLite without mocks, and end-to-end tests of complete pipelines. Use pytest-asyncio for async workflow testing.

Blocking I/O calls inside execute_async() freeze the event loop. Use async libraries like aiohttp instead of requests, await all coroutines, and always handle exceptions to prevent hanging executions.