[FEATURE] Support for Custom Node Types in Graph Pattern

[moritalous]

Problem Statement

Currently, the Graph pattern in Strands Agents only supports Agent or MultiAgentBase instances as nodes. This limitation makes it difficult to implement deterministic business logic, complex preprocessing/postprocessing, and guaranteed execution flows that are essential for production applications.

Proposed Solution

Add support for custom node types that allow:

1. Custom Function Nodes: Ability to add Python functions as nodes for deterministic processing
2. Hybrid Nodes: Nodes that combine deterministic logic with agent capabilities
3. Conditional Logic Nodes: Nodes that perform branching logic without LLM involvement

This would enable developers to create graphs that combine the flexibility of agents with the reliability of deterministic processing, similar to what LangGraph offers.

Use Case

This enhancement would enable:

• Data Processing Pipelines: Reliable preprocessing of data before agent analysis
• Business Rule Implementation: Deterministic business logic execution
• Error Handling: Guaranteed validation and error recovery processes
• Performance Optimization: Avoiding LLM calls for deterministic operations
• Hybrid Workflows: Combining agent creativity with deterministic control

The current limitation forces developers to rely entirely on agent-based processing, where tool execution is not guaranteed and depends on the agent's autonomous decision-making. This creates reliability issues for production applications that require consistent, predictable behavior.

Alternatives Solutions

Currently, developers must work around this limitation by:

• Implementing all logic within agent system prompts (unreliable)
• Using complex tool combinations (execution not guaranteed)
• Building separate processing pipelines outside of Strands Agents

Additional Context

This feature would bring Strands Agents closer to the flexibility offered by frameworks like LangGraph, while maintaining the unique agent-centric approach that makes Strands Agents valuable. The enhancement should maintain full backward compatibility with existing Graph implementations.

[mkmeral]

The Strands Agents SDK already supports custom node types through the MultiAgentBase interface. Below you can see how to implement the requested functionality. Let me know if you think some of these require more work, or rather more documentation.

✅ Requested Features Implementation

1. Custom Function Nodes - Python functions as deterministic graph nodes

class FunctionNode(MultiAgentBase):
    def __init__(self, func, name: str = None):
        super().__init__()
        self.func = func
        self.name = name or func.__name__
        
    async def invoke_async(self, task, **kwargs):
        # Execute deterministic function
        result = self.func(task)
        
        # Return proper AgentResult
        return MultiAgentResult(
            status=Status.COMPLETED,
            results={self.name: NodeResult(result=AgentResult(...))},
            # ... metrics and timing
        )

# Usage
validator = FunctionNode(func=validate_data, name="validator")
builder.add_node(validator, "validator")

2. Hybrid Nodes - Combine deterministic logic with agent capabilities

class HybridNode(MultiAgentBase):
    def __init__(self, agent: Agent, preprocess_func=None, postprocess_func=None):
        self.agent = agent
        self.preprocess_func = preprocess_func
        self.postprocess_func = postprocess_func
        
    async def invoke_async(self, task, **kwargs):
        # Step 1: Deterministic preprocessing
        if self.preprocess_func:
            task = self.preprocess_func(task)
        
        # Step 2: AI agent processing
        agent_result = await self.agent.invoke_async(task)
        
        # Step 3: Deterministic postprocessing
        if self.postprocess_func:
            processed_text = self.postprocess_func(str(agent_result))
            # Create new AgentResult with processed content
            
        return MultiAgentResult(...)

# Usage
ai_processor = HybridNode(
    agent=analyzer,
    preprocess_func=clean_data,
    postprocess_func=format_output
)

3. Conditional Logic Nodes - Deterministic branching without LLM involvement

class ConditionalNode(MultiAgentBase):
    def __init__(self, condition_func, true_result: str, false_result: str):
        self.condition_func = condition_func
        self.true_result = true_result
        self.false_result = false_result
        
    async def invoke_async(self, task, **kwargs):
        # Evaluate condition deterministically
        condition_result = self.condition_func(task)
        result = self.true_result if condition_result else self.false_result
        
        return MultiAgentResult(
            results={self.name: NodeResult(result=AgentResult(...))},
            # ... include condition metadata in state
        )

# Usage
router = ConditionalNode(
    condition_func=lambda x: len(x) > 100,
    true_result="Route to complex processing",
    false_result="Route to simple processing"
)

🚀 Complete Example

from strands import Agent
from strands.multiagent import GraphBuilder

# Create business functions

def validate_input(data: str) -> str:
    """Data Processing Pipeline: Reliable preprocessing before agent analysis."""
    if not data or len(data.strip()) == 0:
        raise ValueError("Empty input data")
    
    if len(data) < 10:
        raise ValueError("Input too short for processing")
    
    return f"✅ Validated: {data[:50]}{'...' if len(data) > 50 else ''}"


def extract_key_info(data: str) -> str:
    """Business Rule Implementation: Deterministic information extraction."""
    # Simulate extracting key business data
    word_count = len(data.split())
    char_count = len(data)
    
    return f"📊 Extracted: {word_count} words, {char_count} characters"


def is_complex_request(data: str) -> bool:
    """Performance Optimization: Route simple requests away from expensive LLM calls."""
    return len(data.split()) > 20 or "complex" in data.lower() or "analysis" in data.lower()


def preprocess_for_ai(data: str) -> str:
    """Hybrid Workflow: Prepare data for AI processing."""
    cleaned = data.strip().replace('\n', ' ').replace('\t', ' ')
    return f"Please analyze this processed data: {cleaned}"


def format_final_output(ai_result: str) -> str:
    """Hybrid Workflow: Format AI output for business consumption."""
    return f"""
📋 ANALYSIS REPORT
==================
{ai_result}
==================
Generated: {time.strftime('%Y-%m-%d %H:%M:%S')}
    """.strip()

# Create custom nodes
validator = FunctionNode(func=validate_input, name="validator")
extractor = FunctionNode(func=extract_key_info, name="extractor")
router = ConditionalNode(
    condition_func=is_complex_request,
    true_result="Complex processing needed",
    false_result="Simple processing sufficient"
)
ai_processor = HybridNode(
    agent=Agent(system_prompt="You are a data analyst"),
    preprocess_func=prepare_for_ai,
    postprocess_func=format_results
)

# Build graph with custom nodes
builder = GraphBuilder()
builder.add_node(validator, "validator")
builder.add_node(extractor, "extractor")
builder.add_node(router, "router")
builder.add_node(ai_processor, "ai_processor")

# Define flow
builder.add_edge("validator", "extractor")
builder.add_edge("extractor", "router")
builder.add_edge("router", "ai_processor")

# Execute
graph = builder.build()
result = await graph.invoke_async("Your data here")

📋 Use Cases Addressed

✅ Data Processing Pipelines: Reliable preprocessing before agent analysis
✅ Business Rule Implementation: Deterministic business logic execution
✅ Error Handling: Guaranteed validation and error recovery
✅ Performance Optimization: Avoid LLM calls for deterministic operations
✅ Hybrid Workflows: Combine agent creativity with deterministic control

[moritalous]

@mkmeral
Thank you very much! I also really appreciate the documentation update.

It's in Japanese, but I wrote a blog post about this topic:
https://qiita.com/moritalous/items/274ea58bce3ccef1589b

Strands Agents and Bedrock AgentCore are gaining a lot of traction in Japan, so I believe this will help drive even more engagement.

[dbschmigelski]

Hi @moritalous, thanks for the question. It seems like @mkmeral was able to guide you to the MultiAgentBase class. Going to close out, but if you have any questions feel free to reopen or create a new issue, thanks