diff --git a/README.md b/README.md
index 207e9258..6f19a4e3 100644
--- a/README.md
+++ b/README.md
@@ -57,6 +57,7 @@ Flo AI is a Python framework that makes building production-ready AI agents and
   - [Create a Tool-Using Agent](#create-a-tool-using-agent)
   - [Create an Agent with Structured Output](#create-an-agent-with-structured-output)
 - [📝 YAML Configuration](#-yaml-configuration)
+- [🔧 Variables System](#-variables-system)
 - [🛠️ Tools](#️-tools)
   - [🎯 @flo_tool Decorator](#-flo_tool-decorator)
 - [🧠 Reasoning Patterns](#-reasoning-patterns)
@@ -269,6 +270,261 @@ agent: Agent = builder.build()
 result: Any = await agent.run(email_thread)
 ```
 
+## 🔧 Variables System
+
+Flo AI supports dynamic variable resolution in agent prompts and inputs using `<variable_name>` syntax. Variables are automatically discovered, validated at runtime, and can be shared across multi-agent workflows.
+
+### ✨ Key Features
+
+- **🔍 Automatic Discovery**: Variables are extracted from system prompts and inputs at runtime
+- **✅ Runtime Validation**: Missing variables are reported with detailed error messages  
+- **🤝 Multi-Agent Support**: Variables can be shared across agent workflows
+- **🛡️ JSON-Safe Syntax**: `<variable>` format avoids conflicts with JSON content
+
+### Basic Usage
+
+```python
+import asyncio
+from typing import Any, Dict
+from flo_ai.builder.agent_builder import AgentBuilder
+from flo_ai.llm import OpenAI
+from flo_ai.models.agent import Agent
+
+async def main() -> None:
+    # Create agent with variables in system prompt
+    agent: Agent = (
+        AgentBuilder()
+        .with_name('Data Analyst')
+        .with_prompt('Analyze <dataset_path> and focus on <key_metric>. Generate insights for <target_audience>.')
+        .with_llm(OpenAI(model='gpt-4o-mini'))
+        .build()
+    )
+    
+    # Define variables at runtime
+    variables: Dict[str, str] = {
+        'dataset_path': '/data/sales_q4_2024.csv',
+        'key_metric': 'revenue growth',
+        'target_audience': 'executive team'
+    }
+    
+    # Run agent with variable resolution
+    result: Any = await agent.run(
+        'Please provide a comprehensive analysis with actionable recommendations.',
+        variables=variables
+    )
+    
+    print(f'Analysis: {result}')
+
+asyncio.run(main())
+```
+
+### Variables in User Input
+
+Variables can also be used in the user input messages:
+
+```python
+import asyncio
+from typing import Any, Dict
+from flo_ai.models.agent import Agent
+from flo_ai.llm import OpenAI
+
+async def input_variables_example() -> None:
+    agent: Agent = Agent(
+        name='content_creator',
+        system_prompt='You are a content creator specializing in <content_type>.',
+        llm=OpenAI(model='gpt-4o-mini')
+    )
+    
+    variables: Dict[str, str] = {
+        'content_type': 'technical blog posts',
+        'topic': 'machine learning fundamentals',
+        'word_count': '1500',
+        'target_level': 'intermediate'
+    }
+    
+    # Variables in both system prompt and user input
+    result: Any = await agent.run(
+        'Create a <word_count>-word article about <topic> for <target_level> readers.',
+        variables=variables
+    )
+    
+    print(f'Content: {result}')
+
+asyncio.run(input_variables_example())
+```
+
+### Multi-Agent Variable Sharing
+
+Variables can be shared and passed between agents in workflows:
+
+```python
+import asyncio
+from typing import Any, Dict, List
+from flo_ai.arium import AriumBuilder
+from flo_ai.models.agent import Agent
+from flo_ai.llm import OpenAI
+
+async def multi_agent_variables() -> List[Any]:
+    llm: OpenAI = OpenAI(model='gpt-4o-mini')
+    
+    # Agent 1: Research phase
+    researcher: Agent = Agent(
+        name='researcher',
+        system_prompt='Research <research_topic> and focus on <research_depth> analysis.',
+        llm=llm
+    )
+    
+    # Agent 2: Writing phase  
+    writer: Agent = Agent(
+        name='writer',
+        system_prompt='Write a <document_type> based on the research for <target_audience>.',
+        llm=llm
+    )
+    
+    # Agent 3: Review phase
+    reviewer: Agent = Agent(
+        name='reviewer',
+        system_prompt='Review the <document_type> for <review_criteria> and provide feedback.',
+        llm=llm
+    )
+    
+    # Shared variables across all agents
+    shared_variables: Dict[str, str] = {
+        'research_topic': 'sustainable energy solutions',
+        'research_depth': 'comprehensive',
+        'document_type': 'white paper',
+        'target_audience': 'policy makers',
+        'review_criteria': 'accuracy and policy relevance'
+    }
+    
+    # Run multi-agent workflow with shared variables
+    result: List[Any] = await (
+        AriumBuilder()
+        .add_agents([researcher, writer, reviewer])
+        .start_with(researcher)
+        .connect(researcher, writer)
+        .connect(writer, reviewer)
+        .end_with(reviewer)
+        .build_and_run(
+            ['Begin comprehensive research and document creation process'],
+            variables=shared_variables
+        )
+    )
+    
+    return result
+
+asyncio.run(multi_agent_variables())
+```
+
+### YAML Configuration with Variables
+
+Variables work seamlessly with YAML-based agent configuration:
+
+```yaml
+apiVersion: flo/alpha-v1
+kind: FloAgent
+metadata:
+  name: personalized-assistant
+  version: 1.0.0
+  description: "Personalized assistant with variable support"
+agent:
+  name: PersonalizedAssistant
+  kind: llm
+  role: <user_role> assistant specialized in <domain_expertise>
+  model:
+    provider: openai
+    name: gpt-4o-mini
+  settings:
+    temperature: 0.3
+    max_retries: 2
+    reasoning_pattern: DIRECT
+  job: >
+    You are a <user_role> focused on <primary_objective>.
+    Your expertise includes <domain_expertise> and you should
+    tailor responses for <experience_level> users.
+    Always consider <priority_constraints> in your recommendations.
+```
+
+```python
+import asyncio
+from typing import Any, Dict
+from flo_ai.builder.agent_builder import AgentBuilder
+from flo_ai.models.agent import Agent
+
+async def yaml_with_variables() -> None:
+    yaml_config: str = """..."""  # Your YAML configuration
+    
+    # Variables for YAML agent
+    variables: Dict[str, str] = {
+        'user_role': 'data scientist',
+        'domain_expertise': 'machine learning and statistical analysis', 
+        'primary_objective': 'deriving actionable insights from data',
+        'experience_level': 'senior',
+        'priority_constraints': 'computational efficiency and model interpretability'
+    }
+    
+    # Create agent from YAML with variables
+    builder: AgentBuilder = AgentBuilder.from_yaml(yaml_str=yaml_config)
+    agent: Agent = builder.build()
+    
+    result: Any = await agent.run(
+        'Help me design an ML pipeline for <use_case> with <data_constraints>',
+        variables={
+            **variables,
+            'use_case': 'customer churn prediction',
+            'data_constraints': 'limited labeled data'
+        }
+    )
+    
+    print(f'ML Pipeline Advice: {result}')
+
+asyncio.run(yaml_with_variables())
+```
+
+### Error Handling and Validation
+
+The variables system provides comprehensive error reporting for missing or invalid variables:
+
+```python
+import asyncio
+from typing import Any, Dict
+from flo_ai.models.agent import Agent
+from flo_ai.llm import OpenAI
+
+async def variable_validation_example() -> None:
+    agent: Agent = Agent(
+        name='validator_example',
+        system_prompt='Process <required_param> and <another_param> for analysis.',
+        llm=OpenAI(model='gpt-4o-mini')
+    )
+    
+    # Incomplete variables (missing 'another_param')
+    incomplete_variables: Dict[str, str] = {
+        'required_param': 'dataset.csv'
+        # 'another_param' is missing
+    }
+    
+    try:
+        result: Any = await agent.run(
+            'Analyze the data in <data_source>',
+            variables=incomplete_variables  # Missing 'another_param' and 'data_source'
+        )
+    except ValueError as e:
+        print(f'Variable validation error: {e}')
+        # Error will list all missing variables with their locations
+
+asyncio.run(variable_validation_example())
+```
+
+### Best Practices
+
+1. **Descriptive Variable Names**: Use clear, descriptive names like `<target_audience>` instead of `<ta>`
+2. **Consistent Naming**: Use consistent variable names across related agents and workflows
+3. **Validation**: Always test your variable resolution before production deployment
+4. **Documentation**: Document expected variables in your agent configurations
+
+The variables system makes Flo AI agents highly reusable and configurable, enabling you to create flexible AI workflows that adapt to different contexts and requirements.
+
 ## 🛠️ Tools
 
 Create custom tools easily with async support:
diff --git a/flo_ai/examples/variables_workflow_example.py b/flo_ai/examples/variables_workflow_example.py
new file mode 100644
index 00000000..4b8b0609
--- /dev/null
+++ b/flo_ai/examples/variables_workflow_example.py
@@ -0,0 +1,242 @@
+"""
+Simple variables workflow example with separate functions for complete and incomplete variables.
+
+This example shows:
+1. Simple function to test complete variables (success case)
+2. Simple function to test incomplete variables (error case)
+"""
+
+import os
+import asyncio
+from flo_ai.builder.agent_builder import AgentBuilder
+from flo_ai.arium.builder import AriumBuilder
+from flo_ai.models.agent import Agent
+from flo_ai.llm.gemini_llm import Gemini
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+api_key = os.getenv('GOOGLE_API_KEY')
+
+
+async def test_single_agent_complete_variables():
+    """Test single agent with complete variables"""
+    print('\n🟢 TESTING SINGLE AGENT - COMPLETE VARIABLES')
+    print('-' * 50)
+
+    llm = Gemini(model='gemini-2.5-flash', temperature=0.7, api_key=api_key)
+
+    # Create agent that uses variables (automatically extracted)
+    translator = (
+        AgentBuilder()
+        .with_name('translator')
+        .with_prompt('You are a translator. Use this tone: <tone>')
+        .with_llm(llm)
+        .build()
+    )
+
+    # Complete variables
+    variables = {
+        'target_language': 'Spanish',
+        'tone': 'formal',
+        'text_to_translate': 'Welcome to our application',
+    }
+
+    print('Variables provided:')
+    for key, value in variables.items():
+        print(f'  ✓ {key}: {value}')
+
+    try:
+        result = await translator.run(
+            'Translate the following text: <text_to_translate> to <target_language>',
+            variables=variables,
+        )
+        print('\n✅ SUCCESS: Single agent executed successfully!')
+        print(f'Result: {result}')
+    except Exception as e:
+        print(f'❌ Unexpected error: {e}')
+
+
+async def test_single_agent_incomplete_variables():
+    """Test single agent with incomplete variables"""
+    print('\n🔴 TESTING SINGLE AGENT - INCOMPLETE VARIABLES')
+    print('-' * 50)
+
+    llm = Gemini(model='gemini-2.5-flash', temperature=0.7, api_key=api_key)
+
+    # Create agent that uses variables (automatically extracted)
+    calculator = (
+        AgentBuilder()
+        .with_name('calculator')
+        .with_prompt('You are a calculator.  Show the result in <format> format.')
+        .with_llm(llm)
+        .build()
+    )
+
+    # Incomplete variables - missing some required variables
+    incomplete_variables = {
+        'operation': 'addition',
+        'number1': '15',
+        # Missing: 'number2' and 'format'
+    }
+
+    print('Variables provided:')
+    for key, value in incomplete_variables.items():
+        print(f'  ✓ {key}: {value}')
+
+    print('\nMissing variables:')
+    print('  ❌ number2')
+    print('  ❌ format')
+
+    try:
+        await calculator.run(
+            'Please calculate the result of <operation> of <number1> and <number2>.',
+            variables=incomplete_variables,
+        )
+        print("❌ ERROR: Should have failed but didn't!")
+    except ValueError as e:
+        print('\n✅ SUCCESS: Expected error caught!')
+        print(f'Error message: {e}')
+
+
+async def test_multi_agent_complete_variables():
+    """Test workflow with complete variables - should succeed"""
+    print('🟢 TESTING MULTI AGENT - COMPLETE VARIABLES')
+    print('-' * 40)
+
+    llm = Gemini(model='gemini-2.5-flash', temperature=0.7, api_key=api_key)
+
+    # Agent 1: Content Creator - uses 'topic' and 'style' variables
+    content_creator = Agent(
+        name='content_creator',
+        system_prompt='You are a content creator. Write in a <style> style. Make it engaging and informative.',
+        llm=llm,
+    )
+
+    # Agent 2: Editor - uses 'target_audience' and 'word_limit' variables
+    editor = Agent(
+        name='editor',
+        system_prompt='You are an editor. Review and refine the content for <target_audience> audience.',
+        llm=llm,
+    )
+
+    # Complete variables - all required variables provided
+    complete_variables = {
+        # Variables for content_creator
+        'topic': 'artificial intelligence trends',
+        'style': 'professional',
+        # Variables for editor
+        'target_audience': 'business executives',
+        'word_limit': '200',
+    }
+
+    print('Variables provided:')
+    for key, value in complete_variables.items():
+        print(f'  ✓ {key}: {value}')
+
+    try:
+        result = await (
+            AriumBuilder()
+            .add_agents([content_creator, editor])
+            .start_with(content_creator)
+            .connect(content_creator, editor)
+            .end_with(editor)
+            .build_and_run(
+                inputs=[
+                    'Create content about the given topic: <topic>. Keeping it under <word_limit> words.'
+                ],
+                variables=complete_variables,
+            )
+        )
+
+        print('\n✅ SUCCESS: Workflow completed with complete variables!')
+        print('Final result:')
+        for i, message in enumerate(result, 1):
+            if isinstance(message, str):
+                print(f'  {i}. {message}...')
+            else:
+                print(f'  {i}. {message}')
+
+    except Exception as e:
+        print(f'❌ Unexpected error: {e}')
+
+
+async def test_multi_agent_incomplete_variables():
+    """Test workflow with incomplete variables - should fail with clear error"""
+    print('\n🔴 TESTING MULTI AGENT - INCOMPLETE VARIABLES')
+    print('-' * 40)
+
+    llm = Gemini(model='gemini-2.5-flash', temperature=0.7, api_key=api_key)
+
+    # Agent 1: Researcher - uses 'research_topic' and 'depth_level' variables
+    researcher = Agent(
+        name='researcher',
+        system_prompt='You are a researcher. Research <depth_level> level of detail.',
+        llm=llm,
+    )
+
+    # Agent 2: Summarizer - uses 'summary_format' and 'key_points' variables
+    summarizer = Agent(
+        name='summarizer',
+        system_prompt='You are a summarizer. Create a <summary_format> summary.',
+        llm=llm,
+    )
+
+    # Incomplete variables - missing several required variables
+    incomplete_variables = {
+        'research_topic': 'climate change solutions',
+        'summary_format': 'bullet_point',
+        # Missing: 'depth_level' (for researcher)
+        # Missing: 'key_points' (for summarizer)
+    }
+
+    print('Variables provided:')
+    for key, value in incomplete_variables.items():
+        print(f'  ✓ {key}: {value}')
+
+    print('\nMissing variables:')
+    print('  ❌ depth_level (required by researcher)')
+    print('  ❌ key_points (required by summarizer)')
+
+    try:
+        await (
+            AriumBuilder()
+            .add_agents([researcher, summarizer])
+            .start_with(researcher)
+            .connect(researcher, summarizer)
+            .end_with(summarizer)
+            .build_and_run(
+                inputs=[
+                    'Research the given topic: <research_topic>. Highlight <key_points> key points'
+                ],
+                variables=incomplete_variables,
+            )
+        )
+        print("❌ ERROR: Should have failed but didn't!")
+
+    except ValueError as e:
+        print('\n✅ SUCCESS: Expected error caught!')
+        print(f'Error message: {e}')
+
+
+if __name__ == '__main__':
+    print('Available functions to call:')
+    print(
+        '• asyncio.run(test_single_agent_complete_variables()) - Test single agent complete variables only'
+    )
+    print(
+        '• asyncio.run(test_single_agent_incomplete_variables()) - Test single agent incomplete variables only'
+    )
+    print(
+        '• asyncio.run(test_multi_agent_complete_variables())     - Test multi agent complete variables only'
+    )
+    print(
+        '• asyncio.run(test_multi_agent_incomplete_variables())   - Test multi agent incomplete variables only'
+    )
+    print()
+
+    asyncio.run(test_single_agent_complete_variables())
+    asyncio.run(test_single_agent_incomplete_variables())
+    asyncio.run(test_multi_agent_complete_variables())
+    asyncio.run(test_multi_agent_incomplete_variables())
diff --git a/flo_ai/examples/variables_workflow_yaml_example.py b/flo_ai/examples/variables_workflow_yaml_example.py
new file mode 100644
index 00000000..190f6186
--- /dev/null
+++ b/flo_ai/examples/variables_workflow_yaml_example.py
@@ -0,0 +1,325 @@
+"""
+Simple variables workflow example with YAML agent configurations.
+Demonstrates the same functionality as variables_workflow_example.py but using YAML configs.
+
+This example shows:
+1. Simple function to test complete variables (success case) with YAML agents
+2. Simple function to test incomplete variables (error case) with YAML agents
+3. YAML-based agent configuration instead of direct Agent() instantiation
+"""
+
+import os
+import asyncio
+from flo_ai.builder.agent_builder import AgentBuilder
+from flo_ai.arium.builder import AriumBuilder
+from flo_ai.llm.gemini_llm import Gemini
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+api_key = os.getenv('GOOGLE_API_KEY')
+
+
+async def test_single_agent_complete_variables():
+    """Test single agent with complete variables"""
+    print('\n🟢 TESTING SINGLE AGENT - COMPLETE VARIABLES (YAML)')
+    print('-' * 50)
+
+    llm = Gemini(model='gemini-2.5-flash', temperature=0.7, api_key=api_key)
+
+    # Create agent that uses variables (automatically extracted) with YAML
+    translator_yaml = """
+apiVersion: flo/alpha-v1
+metadata:
+  name: translator-agent
+  version: 1.0.0
+  description: "Agent for translating text with specified tone"
+
+agent:
+  name: translator
+  role: Professional Translator
+  model:
+    provider: gemini
+    name: gemini-2.5-flash
+  settings:
+    temperature: 0.7
+  job: You are a translator. Use this tone <tone>
+"""
+
+    translator = AgentBuilder.from_yaml(yaml_str=translator_yaml).with_llm(llm).build()
+
+    # Complete variables
+    variables = {
+        'target_language': 'Spanish',
+        'tone': 'formal',
+        'text_to_translate': 'Welcome to our application',
+    }
+
+    print('Variables provided:')
+    for key, value in variables.items():
+        print(f'  ✓ {key}: {value}')
+
+    try:
+        result = await translator.run(
+            'Translate the following text: <text_to_translate> to <target_language>',
+            variables=variables,
+        )
+        print('\n✅ SUCCESS: Single agent executed successfully!')
+        print(f'Result: {result}')
+    except Exception as e:
+        print(f'❌ Unexpected error: {e}')
+
+
+async def test_single_agent_incomplete_variables():
+    """Test single agent with incomplete variables"""
+    print('\n🔴 TESTING SINGLE AGENT - INCOMPLETE VARIABLES (YAML)')
+    print('-' * 50)
+
+    llm = Gemini(model='gemini-2.5-flash', temperature=0.7, api_key=api_key)
+
+    # Create agent that uses variables (automatically extracted) with YAML
+    calculator_yaml = """
+apiVersion: flo/alpha-v1
+metadata:
+  name: calculator-agent
+  version: 1.0.0
+  description: "Agent for performing calculations in specified format"
+
+agent:
+  name: calculator
+  role: Mathematical Calculator
+  model:
+    provider: gemini
+    name: gemini-2.5-flash
+  settings:
+    temperature: 0.7
+  job: You are a calculator. Show the result in <format> format.
+"""
+
+    calculator = AgentBuilder.from_yaml(yaml_str=calculator_yaml).with_llm(llm).build()
+
+    # Incomplete variables - missing some required variables
+    incomplete_variables = {
+        'operation': 'addition',
+        'number1': '15',
+        # Missing: 'number2' and 'format'
+    }
+
+    print('Variables provided:')
+    for key, value in incomplete_variables.items():
+        print(f'  ✓ {key}: {value}')
+
+    print('\nMissing variables:')
+    print('  ❌ number2')
+    print('  ❌ format')
+
+    try:
+        await calculator.run(
+            'Please calculate the result of <operation> of <number1> and <number2>.',
+            variables=incomplete_variables,
+        )
+        print("❌ ERROR: Should have failed but didn't!")
+    except ValueError as e:
+        print('\n✅ SUCCESS: Expected error caught!')
+        print(f'Error message: {e}')
+
+
+async def test_multi_agent_complete_variables():
+    """Test workflow with complete variables - should succeed"""
+    print('🟢 TESTING MULTI AGENT - COMPLETE VARIABLES (YAML)')
+    print('-' * 40)
+
+    llm = Gemini(model='gemini-2.5-flash', temperature=0.7, api_key=api_key)
+
+    # Agent 1: Content Creator - uses 'topic' and 'style' variables
+    content_creator_yaml = """
+apiVersion: flo/alpha-v1
+metadata:
+  name: content-creator-agent
+  version: 1.0.0
+  description: "Agent for creating content with specified style"
+
+agent:
+  name: content_creator
+  role: Content Creator
+  model:
+    provider: gemini
+    name: gemini-2.5-flash
+  settings:
+    temperature: 0.7
+  job: You are a content creator. Write in a <style> style. Make it engaging and informative.
+"""
+
+    content_creator = (
+        AgentBuilder.from_yaml(yaml_str=content_creator_yaml).with_llm(llm).build()
+    )
+
+    # Agent 2: Editor - uses 'target_audience' and 'word_limit' variables
+    editor_yaml = """
+apiVersion: flo/alpha-v1
+metadata:
+  name: editor-agent
+  version: 1.0.0
+  description: "Agent for editing content for specific audiences"
+
+agent:
+  name: editor
+  role: Content Editor
+  model:
+    provider: gemini
+    name: gemini-2.5-flash
+  settings:
+    temperature: 0.7
+  job: You are an editor. Review and refine the content for <target_audience> audience.
+"""
+
+    editor = AgentBuilder.from_yaml(yaml_str=editor_yaml).with_llm(llm).build()
+
+    # Complete variables - all required variables provided
+    complete_variables = {
+        # Variables for content_creator
+        'topic': 'artificial intelligence trends',
+        'style': 'professional',
+        # Variables for editor
+        'target_audience': 'business executives',
+        'word_limit': '200',
+    }
+
+    print('Variables provided:')
+    for key, value in complete_variables.items():
+        print(f'  ✓ {key}: {value}')
+
+    try:
+        result = await (
+            AriumBuilder()
+            .add_agents([content_creator, editor])
+            .start_with(content_creator)
+            .connect(content_creator, editor)
+            .end_with(editor)
+            .build_and_run(
+                inputs=[
+                    'Create content about the given topic: <topic>. Keeping it under <word_limit> words.'
+                ],
+                variables=complete_variables,
+            )
+        )
+
+        print('\n✅ SUCCESS: Workflow completed with complete variables!')
+        print('Final result:')
+        for i, message in enumerate(result, 1):
+            if isinstance(message, str):
+                print(f'  {i}. {message}...')
+            else:
+                print(f'  {i}. {message}')
+
+    except Exception as e:
+        print(f'❌ Unexpected error: {e}')
+
+
+async def test_multi_agent_incomplete_variables():
+    """Test workflow with incomplete variables - should fail with clear error"""
+    print('\n🔴 TESTING MULTI AGENT - INCOMPLETE VARIABLES (YAML)')
+    print('-' * 40)
+
+    llm = Gemini(model='gemini-2.5-flash', temperature=0.7, api_key=api_key)
+
+    # Agent 1: Researcher - uses 'research_topic' and 'depth_level' variables
+    researcher_yaml = """
+apiVersion: flo/alpha-v1
+metadata:
+  name: researcher-agent
+  version: 1.0.0
+  description: "Agent for conducting research at specified depth"
+
+agent:
+  name: researcher
+  role: Research Specialist
+  model:
+    provider: gemini
+    name: gemini-2.5-flash
+  settings:
+    temperature: 0.7
+  job: You are a researcher. Research <depth_level> level of detail.
+"""
+
+    researcher = AgentBuilder.from_yaml(yaml_str=researcher_yaml).with_llm(llm).build()
+
+    # Agent 2: Summarizer - uses 'summary_format' and 'key_points' variables
+    summarizer_yaml = """
+apiVersion: flo/alpha-v1
+metadata:
+  name: summarizer-agent
+  version: 1.0.0
+  description: "Agent for creating summaries in specified format"
+
+agent:
+  name: summarizer
+  role: Content Summarizer
+  model:
+    provider: gemini
+    name: gemini-2.5-flash
+  settings:
+    temperature: 0.7
+  job: You are a summarizer. Create a <summary_format> summary.
+"""
+
+    summarizer = AgentBuilder.from_yaml(yaml_str=summarizer_yaml).with_llm(llm).build()
+
+    # Incomplete variables - missing several required variables
+    incomplete_variables = {
+        'research_topic': 'climate change solutions',
+        'summary_format': 'bullet_point',
+        # Missing: 'depth_level' (for researcher)
+        # Missing: 'key_points' (for summarizer)
+    }
+
+    print('Variables provided:')
+    for key, value in incomplete_variables.items():
+        print(f'  ✓ {key}: {value}')
+
+    print('\nMissing variables:')
+    print('  ❌ depth_level (required by researcher)')
+    print('  ❌ key_points (required by summarizer)')
+
+    try:
+        await (
+            AriumBuilder()
+            .add_agents([researcher, summarizer])
+            .start_with(researcher)
+            .connect(researcher, summarizer)
+            .end_with(summarizer)
+            .build_and_run(
+                inputs=[
+                    'Research the given topic: <research_topic>. Highlight <key_points> key points'
+                ],
+                variables=incomplete_variables,
+            )
+        )
+        print("❌ ERROR: Should have failed but didn't!")
+
+    except ValueError as e:
+        print('\n✅ SUCCESS: Expected error caught!')
+        print(f'Error message: {e}')
+
+
+if __name__ == '__main__':
+    print('Available functions to call:')
+    print(
+        '• asyncio.run(test_single_agent_complete_variables()) - Test single agent complete variables only'
+    )
+    print(
+        '• asyncio.run(test_single_agent_incomplete_variables()) - Test single agent incomplete variables only'
+    )
+    print(
+        '• asyncio.run(test_multi_agent_complete_variables())     - Test multi agent complete variables only'
+    )
+    print(
+        '• asyncio.run(test_multi_agent_incomplete_variables())   - Test multi agent incomplete variables only'
+    )
+    print()
+
+    asyncio.run(test_single_agent_complete_variables())
+    asyncio.run(test_single_agent_incomplete_variables())
+    asyncio.run(test_multi_agent_complete_variables())
+    asyncio.run(test_multi_agent_incomplete_variables())
diff --git a/flo_ai/flo_ai/arium/arium.py b/flo_ai/flo_ai/arium/arium.py
index 561d604f..a1a20183 100644
--- a/flo_ai/flo_ai/arium/arium.py
+++ b/flo_ai/flo_ai/arium/arium.py
@@ -1,11 +1,17 @@
 from flo_ai.arium.base import BaseArium
 from flo_ai.arium.memory import MessageMemory, BaseMemory
 from flo_ai.llm.base_llm import ImageMessage
-from typing import List
+from typing import List, Dict, Any, Optional
 from flo_ai.models.agent import Agent
 from flo_ai.tool.base_tool import Tool
 from flo_ai.arium.models import StartNode, EndNode
 from flo_ai.utils.logger import logger
+from flo_ai.utils.variable_extractor import (
+    extract_variables_from_inputs,
+    extract_agent_variables,
+    validate_multi_agent_variables,
+    resolve_variables,
+)
 
 
 class Arium(BaseArium):
@@ -18,7 +24,11 @@ def compile(self):
         self.validate_graph()
         self.is_compiled = True
 
-    async def run(self, inputs: List[str | ImageMessage]):
+    async def run(
+        self,
+        inputs: List[str | ImageMessage],
+        variables: Optional[Dict[str, Any]] = None,
+    ):
         if not self.is_compiled:
             raise ValueError('Arium is not compiled')
 
@@ -28,7 +38,14 @@ async def run(self, inputs: List[str | ImageMessage]):
         if not self.nodes:
             raise ValueError('Arium has no nodes')
 
-        return await self._execute_graph(inputs)
+        # Extract and validate variables from inputs and all agents
+        self._extract_and_validate_variables(inputs, variables)
+
+        # Resolve variables in inputs and agent prompts
+        resolved_inputs = self._resolve_inputs(inputs, variables)
+        self._resolve_agent_prompts(variables)
+
+        return await self._execute_graph(resolved_inputs)
 
     async def _execute_graph(self, inputs: List[str | ImageMessage]):
         [self.memory.add(msg) for msg in inputs]
@@ -59,9 +76,81 @@ async def _execute_graph(self, inputs: List[str | ImageMessage]):
 
         return self.memory.get()
 
+    def _extract_and_validate_variables(
+        self, inputs: List[str | ImageMessage], variables: Dict[str, Any]
+    ) -> None:
+        """Extract variables from inputs and agents, then validate them.
+
+        Args:
+            inputs: List of input messages
+            variables: Dictionary of variable name to value mappings
+
+        Raises:
+            ValueError: If any required variables are missing
+        """
+        # Extract variables from inputs
+        input_variables = extract_variables_from_inputs(inputs)
+
+        # Extract variables from all agents in the workflow
+        agents_variables = {}
+        for node in self.nodes.values():
+            if isinstance(node, Agent):
+                agent_vars = extract_agent_variables(node)
+                if agent_vars:
+                    agents_variables[node.name] = agent_vars
+
+        # Validate input variables separately with cleaner error message
+        if input_variables:
+            missing_input_vars = input_variables - set(variables.keys())
+            if missing_input_vars:
+                provided_keys = sorted(variables.keys())
+                raise ValueError(
+                    f'Input contains missing variables: {sorted(missing_input_vars)}. '
+                    f'Provided variables: {provided_keys}'
+                )
+
+        # Validate agent variables with detailed agent breakdown
+        if agents_variables:
+            validate_multi_agent_variables(agents_variables, variables)
+
+    def _resolve_inputs(
+        self, inputs: List[str | ImageMessage], variables: Dict[str, Any]
+    ) -> List[str | ImageMessage]:
+        """Resolve variables in input messages.
+
+        Args:
+            inputs: List of input messages
+            variables: Dictionary of variable name to value mappings
+
+        Returns:
+            List of inputs with variables resolved
+        """
+        resolved_inputs = []
+        for input_item in inputs:
+            if isinstance(input_item, str):
+                # Resolve variables in text input
+                resolved_input = resolve_variables(input_item, variables)
+                resolved_inputs.append(resolved_input)
+            else:
+                # ImageMessage objects don't need variable resolution
+                resolved_inputs.append(input_item)
+        return resolved_inputs
+
+    def _resolve_agent_prompts(self, variables: Dict[str, Any]) -> None:
+        """Resolve variables in all agent system prompts and mark them as resolved.
+
+        Args:
+            variables: Dictionary of variable name to value mappings
+        """
+        for node in self.nodes.values():
+            if isinstance(node, Agent):
+                node.system_prompt = resolve_variables(node.system_prompt, variables)
+                node.resolved_variables = True
+
     async def _execute_node(self, node: Agent | Tool | StartNode | EndNode):
         if isinstance(node, Agent):
-            return await node.run(self.memory.get())
+            # Variables are already resolved, pass empty dict to avoid re-processing
+            return await node.run(self.memory.get(), variables={})
         elif isinstance(node, Tool):
             return await node.execute(self.memory.get())
         elif isinstance(node, StartNode):
diff --git a/flo_ai/flo_ai/arium/builder.py b/flo_ai/flo_ai/arium/builder.py
index 7511bb2a..b343bcfa 100644
--- a/flo_ai/flo_ai/arium/builder.py
+++ b/flo_ai/flo_ai/arium/builder.py
@@ -1,4 +1,4 @@
-from typing import List, Optional, Callable, Union
+from typing import List, Optional, Callable, Union, Dict, Any
 from flo_ai.arium.arium import Arium
 from flo_ai.arium.memory import MessageMemory, BaseMemory
 from flo_ai.models.agent import Agent
@@ -128,10 +128,14 @@ def build(self) -> Arium:
         self._arium = arium
         return arium
 
-    async def build_and_run(self, inputs: List[Union[str, ImageMessage]]) -> List[dict]:
-        """Build the Arium and run it with the given inputs."""
+    async def build_and_run(
+        self,
+        inputs: List[Union[str, ImageMessage]],
+        variables: Optional[Dict[str, Any]] = None,
+    ) -> List[dict]:
+        """Build the Arium and run it with the given inputs and optional runtime variables."""
         arium = self.build()
-        return await arium.run(inputs)
+        return await arium.run(inputs, variables=variables)
 
     def visualize(
         self, output_path: str = 'arium_graph.png', title: str = 'Arium Workflow'
diff --git a/flo_ai/flo_ai/models/agent.py b/flo_ai/flo_ai/models/agent.py
index fd1cbc84..053d42ee 100644
--- a/flo_ai/flo_ai/models/agent.py
+++ b/flo_ai/flo_ai/models/agent.py
@@ -5,6 +5,12 @@
 from flo_ai.tool.base_tool import Tool, ToolExecutionError
 from flo_ai.models.agent_error import AgentError
 from flo_ai.utils.logger import logger
+from flo_ai.utils.variable_extractor import (
+    extract_variables_from_inputs,
+    extract_agent_variables,
+    validate_multi_agent_variables,
+    resolve_variables,
+)
 
 
 class Agent(BaseAgent):
@@ -40,35 +46,79 @@ def __init__(
         self.output_schema = output_schema
         self.role = role
 
-    async def run(self, inputs: List[str | ImageMessage] | str) -> str:
+    async def run(
+        self,
+        inputs: List[str | ImageMessage] | str,
+        variables: Optional[Dict[str, Any]] = None,
+    ) -> str:
+        variables = variables or {}
+
         if isinstance(inputs, str):
             inputs = [inputs]
 
-        for input in inputs:
-            if isinstance(input, ImageMessage):
-                self.add_to_history('user', self.llm.format_image_in_message(input))
-            else:
-                self.add_to_history('user', input)
+        # Perform runtime variable validation if not already resolved (single agent usage)
+        if not self.resolved_variables:
+            # Extract variables from inputs and system prompt
+            input_variables = extract_variables_from_inputs(inputs)
+            agent_variables = extract_agent_variables(self)
+            all_required_variables = input_variables.union(agent_variables)
+
+            # Validate that all required variables are provided
+            if all_required_variables:
+                agents_variables = {self.name: all_required_variables}
+                validate_multi_agent_variables(agents_variables, variables)
+
+            # Resolve variables and mark as resolved
+            self.system_prompt = resolve_variables(self.system_prompt, variables)
+
+            # Process inputs and resolve variables in string inputs
+            for input in inputs:
+                if isinstance(input, ImageMessage):
+                    self.add_to_history('user', self.llm.format_image_in_message(input))
+                else:
+                    # Resolve variables in text input
+                    resolved_input = resolve_variables(input, variables)
+                    self.add_to_history('user', resolved_input)
+
+            # after resolving agent system prompts and inputs, mark variables as resolved
+            self.resolved_variables = True
+
+        else:
+            # Variables already resolved, process inputs without variable resolution
+            for input in inputs:
+                if isinstance(input, ImageMessage):
+                    self.add_to_history('user', self.llm.format_image_in_message(input))
+                else:
+                    self.add_to_history('user', input)
 
         retry_count = 0
 
         # If no tools, act as conversational agent
         if not self.tools:
-            return await self._run_conversational(retry_count)
+            return await self._run_conversational(retry_count, variables)
 
         # Otherwise, run as tool agent
-        return await self._run_with_tools(retry_count)
+        return await self._run_with_tools(retry_count, variables)
 
-    async def _run_conversational(self, retry_count: int) -> str:
+    async def _run_conversational(
+        self, retry_count: int, variables: Optional[Dict[str, Any]] = None
+    ) -> str:
         """Run as a conversational agent when no tools are provided"""
+        variables = variables or {}
+
         while retry_count < self.max_retries:
             try:
+                # Resolve variables in system prompt
+                system_content = (
+                    self._get_cot_prompt(variables)
+                    if self.reasoning_pattern == ReasoningPattern.COT
+                    else resolve_variables(self.system_prompt, variables)
+                )
+
                 messages = [
                     {
                         'role': 'system',
-                        'content': self._get_cot_prompt()
-                        if self.reasoning_pattern == ReasoningPattern.COT
-                        else self.system_prompt,
+                        'content': system_content,
                     }
                 ] + self.conversation_history
 
@@ -111,18 +161,26 @@ async def _run_conversational(self, retry_count: int) -> str:
                         original_error=e,
                     )
 
-    async def _run_with_tools(self, retry_count: int = 0) -> str:
+    async def _run_with_tools(
+        self, retry_count: int = 0, variables: Optional[Dict[str, Any]] = None
+    ) -> str:
         """Run as a tool-using agent when tools are provided"""
+        variables = variables or {}
+
         while retry_count < self.max_retries:
             try:
+                # Resolve variables in system prompt based on reasoning pattern
+                if self.reasoning_pattern == ReasoningPattern.REACT:
+                    system_content = self._get_react_prompt(variables)
+                elif self.reasoning_pattern == ReasoningPattern.COT:
+                    system_content = self._get_cot_prompt(variables)
+                else:
+                    system_content = resolve_variables(self.system_prompt, variables)
+
                 messages = [
                     {
                         'role': 'system',
-                        'content': self._get_react_prompt()
-                        if self.reasoning_pattern == ReasoningPattern.REACT
-                        else self._get_cot_prompt()
-                        if self.reasoning_pattern == ReasoningPattern.COT
-                        else self.system_prompt,
+                        'content': system_content,
                     }
                 ] + self.conversation_history
 
@@ -239,12 +297,18 @@ async def _run_with_tools(self, retry_count: int = 0) -> str:
 
         raise AgentError(f'Failed after maximum {self.max_retries} attempts.')
 
-    def _get_react_prompt(self) -> str:
+    def _get_react_prompt(self, variables: Optional[Dict[str, Any]] = None) -> str:
         """Get system prompt modified for ReACT pattern"""
+        variables = variables or {}
+
         tools_desc = '\n'.join(
             [f'- {tool.name}: {tool.description}' for tool in self.tools]
         )
-        react_prompt = f"""{self.system_prompt}
+
+        # Resolve variables in the base system prompt
+        resolved_system_prompt = resolve_variables(self.system_prompt, variables)
+
+        react_prompt = f"""{resolved_system_prompt}
             When solving tasks, follow this format:
 
             Thought: Analyze the situation and think about what to do
@@ -263,12 +327,18 @@ def _get_react_prompt(self) -> str:
 
         return react_prompt
 
-    def _get_cot_prompt(self) -> str:
+    def _get_cot_prompt(self, variables: Optional[Dict[str, Any]] = None) -> str:
         """Get system prompt modified for Chain of Thought pattern"""
+        variables = variables or {}
+
         tools_desc = '\n'.join(
             [f'- {tool.name}: {tool.description}' for tool in self.tools]
         )
-        cot_prompt = f"""{self.system_prompt}
+
+        # Resolve variables in the base system prompt
+        resolved_system_prompt = resolve_variables(self.system_prompt, variables)
+
+        cot_prompt = f"""{resolved_system_prompt}
             When solving tasks, follow this Chain of Thought reasoning format:
 
             Let me think through this step by step:
diff --git a/flo_ai/flo_ai/models/base_agent.py b/flo_ai/flo_ai/models/base_agent.py
index 106c2788..ab506b46 100644
--- a/flo_ai/flo_ai/models/base_agent.py
+++ b/flo_ai/flo_ai/models/base_agent.py
@@ -29,6 +29,7 @@ def __init__(
         self.agent_type = agent_type
         self.llm = llm
         self.max_retries = max_retries
+        self.resolved_variables = False
         self.conversation_history: List[Dict[str, str]] = []
 
     @abstractmethod
diff --git a/flo_ai/flo_ai/utils/variable_extractor.py b/flo_ai/flo_ai/utils/variable_extractor.py
new file mode 100644
index 00000000..3d3f7bba
--- /dev/null
+++ b/flo_ai/flo_ai/utils/variable_extractor.py
@@ -0,0 +1,157 @@
+"""
+Variable extraction utilities for the Flo AI framework.
+
+This module provides functions to extract variable placeholders from text,
+inputs, and agent configurations for runtime variable validation.
+"""
+
+import re
+from typing import List, Set, Dict, Any
+from flo_ai.llm.base_llm import ImageMessage
+
+
+def extract_variables_from_text(text: str) -> Set[str]:
+    """Extract variable placeholders from text using <variable_name> pattern.
+
+    Args:
+        text: Text containing variable placeholders like <variable_name>
+
+    Returns:
+        Set of variable names found in the text
+
+    Examples:
+        >>> extract_variables_from_text("Hello <name>, your <role> is important")
+        {'name', 'role'}
+        >>> extract_variables_from_text("No variables here")
+        set()
+    """
+    if not text:
+        return set()
+
+    # Use regex to find all <variable_name> patterns
+    # \w+ matches word characters (letters, digits, underscore)
+    variable_pattern = r'<(\w+)>'
+    matches = re.findall(variable_pattern, text)
+    return set(matches)
+
+
+def extract_variables_from_inputs(inputs: List[str | ImageMessage]) -> Set[str]:
+    """Extract variable placeholders from a list of input messages.
+
+    Args:
+        inputs: List of input strings or ImageMessages
+
+    Returns:
+        Set of variable names found across all string inputs
+
+    Note:
+        ImageMessage objects are skipped as they don't contain variable placeholders
+    """
+    all_variables = set()
+
+    for input_item in inputs:
+        if isinstance(input_item, str):
+            variables = extract_variables_from_text(input_item)
+            all_variables.update(variables)
+        # Skip ImageMessage objects as they don't contain text variables
+
+    return all_variables
+
+
+def extract_agent_variables(agent) -> Set[str]:
+    """Extract variable placeholders from an agent's system prompt.
+
+    Args:
+        agent: Agent instance with a system_prompt attribute
+
+    Returns:
+        Set of variable names found in the agent's system prompt
+
+    Note:
+        This function avoids importing Agent to prevent circular imports
+    """
+    if not hasattr(agent, 'system_prompt'):
+        return set()
+
+    return extract_variables_from_text(agent.system_prompt)
+
+
+def validate_variables(
+    required_variables: Set[str], provided_variables: Dict[str, Any], context: str = ''
+) -> None:
+    """Validate that all required variables are provided.
+
+    Args:
+        required_variables: Set of variable names that are required
+        provided_variables: Dictionary of variable name to value mappings
+        context: Optional context string for error messages (e.g., agent name)
+
+    Raises:
+        ValueError: If any required variables are missing
+    """
+    missing = required_variables - set(provided_variables.keys())
+    if missing:
+        available = list(provided_variables.keys()) if provided_variables else []
+        context_str = f' for {context}' if context else ''
+        raise ValueError(
+            f'Missing required variables{context_str}: {sorted(missing)}. '
+            f'Available variables: {available}'
+        )
+
+
+def resolve_variables(text: str, variables: Dict[str, Any]) -> str:
+    """Replace <variable_name> patterns with actual values
+
+    Args:
+        text: Text containing variable placeholders like <variable_name>
+        variables: Dictionary of variable name to value mappings
+
+    Returns:
+        Text with variables resolved
+
+    Raises:
+        ValueError: If a variable placeholder is found but not provided in variables
+    """
+    if not text or not variables:
+        return text
+
+    def replace_var(match):
+        var_name = match.group(1)
+        if var_name in variables:
+            return str(variables[var_name])
+        else:
+            available = list(variables.keys())
+            raise ValueError(
+                f"Variable '{var_name}' referenced in text but not provided. "
+                f'Available variables: {available}'
+            )
+
+    return re.sub(r'<(\w+)>', replace_var, text)
+
+
+def validate_multi_agent_variables(
+    agents_variables: Dict[str, Set[str]], provided_variables: Dict[str, Any]
+) -> None:
+    """Validate variables for multiple agents and provide detailed error messages.
+
+    Args:
+        agents_variables: Dictionary mapping agent names to their required variables
+        provided_variables: Dictionary of variable name to value mappings
+
+    Raises:
+        ValueError: If any agents are missing required variables, with detailed breakdown
+    """
+    missing_by_agent = {}
+    provided_keys = set(provided_variables.keys())
+
+    for agent_name, required_vars in agents_variables.items():
+        missing = required_vars - provided_keys
+        if missing:
+            missing_by_agent[agent_name] = sorted(missing)
+
+    if missing_by_agent:
+        error_msg = 'Missing required variables for agents:\n'
+        for agent_name, missing_vars in missing_by_agent.items():
+            error_msg += f"  - Agent '{agent_name}': {missing_vars}\n"
+        error_msg += f'Provided variables: {sorted(provided_keys)}'
+        raise ValueError(error_msg)