Reactive Agents

🚀 Reactive AI Agent Framework

[](https://github.com/tylerjrbuell/reactive-agents/actions/workflows/ci.yml) [](https://badge.fury.io/py/reactive-agents) [](https://pypi.org/project/reactive-agents/) [](https://opensource.org/licenses/MIT) [](https://codecov.io/gh/tylerjrbuell/reactive-agents) [](https://github.com/psf/black) [](https://pepy.tech/project/reactive-agents) [](https://github.com/tylerjrbuell/reactive-agents/stargazers)

[](https://tylerjrbuell.github.io/reactive-agents/) [](https://discord.gg/WVxTnHt8)

An Elegant, Powerful, and Flexible Framework for Building Reactive AI Agents

[🏁 Quick Start](#quick-start) • 📖 Documentation • [🎯 Features](#features) • [🛠️ Installation](#installation) • [💡 Examples](#examples) • [🤝 Contributing](#contributing)

---

🌟 What is Reactive Agents?

Reactive Agents is a cutting-edge AI agent framework that makes building intelligent, autonomous agents as simple as Laravel makes web development. With its elegant builder pattern, comprehensive tooling ecosystem, and production-ready architecture, you can create sophisticated AI agents that think, plan, execute, and adapt.

🔎 Definition — "Reactive" (adj.)

reactive /ˈriːæk.tɪv/

1. Promptly responsive to change or external stimuli; able to sense, interpret, and act in real time.
2. Architected for rapid feedback loops, context-aware adaptation, and low-latency decision-making.

🚀 Why "Reactive"?

Reactive agents turn sensing into instant value — they detect shifts, call the right tools, and adjust plans on the fly. That means faster answers, fewer failures, better user experiences, and systems that scale gracefully under real-world uncertainty. In short: reactive = faster, smarter, and more reliable AI that drives reliable outcomes now.

🎯 Perfect For

• 🔬 Research Automation - Intelligent web research and data analysis
• 📊 Business Intelligence - Automated reporting and decision support
• 🛠️ DevOps & Infrastructure - Intelligent monitoring and automation
• 💬 Customer Support - Smart assistants with tool integration
• 📈 Data Processing - Complex workflows with multiple data sources
• 🎮 Interactive Applications - AI-powered user experiences
• 🤖 Multi-Agent Systems - Orchestrated AI teams solving complex problems
• ⚙️ Automation & Scripting - Intelligent task automation

✨ Key Features

🧠 Multiple Reasoning Strategies

Composable strategies with component-based architecture: Strategies are modular and pluggable, built from discrete components (planners, executors, reflectors, and goal evaluators) that you can mix-and-match to craft custom reasoning flows.

• Modular components — planners, executors, reflectors, and evaluators are independent and swappable.
• Pluggable strategies — implement BaseReasoningStrategy and register with StrategyManager to add new strategies.
• Testable & reusable — small, well-typed components make unit testing and reuse simple.
• Designed for composition — use the Adaptive strategy or compose multiple strategies to handle complex, dynamic tasks.

Pre-built strategies include:

• Reactive: Fast, direct problem-solving
• Plan-Execute-Reflect: Structured approach for complex tasks
• Reflect-Decide-Act: Adaptive strategy for dynamic environments
• Adaptive: AI-driven strategy selection based on task complexity

🔧 Comprehensive Tool Ecosystem

• Custom Python Tools with @tool() decorator
• Model Context Protocol (MCP) integration
• Pre-built Tools: Web search, file operations, databases, and more
• Tool Composition and validation system

🏗️ Production-Ready Architecture

• Event-Driven Design with real-time monitoring
• Robust Error Recovery with intelligent retry mechanisms
• Memory Management with vector storage and persistence
• Performance Monitoring with detailed metrics and scoring
• Context Optimization with adaptive pruning strategies

🔄 Advanced Workflow Management

• Multi-Agent Orchestration with dependency management
• A2A Communication (Agent-to-Agent) protocols
• Parallel Execution and synchronization
• Workflow Templates for common patterns

🎛️ Developer Experience

• Fluent Builder API with sensible defaults
• Type Safety with Pydantic models throughout
• Comprehensive Logging with structured events
• 🚧 Plugin System for extensibility
• Hot-reloading for development workflows

---

🏁 Quick Start

Installation

pip install reactive-agents

Your First Agent (30 seconds)

import asyncio
from reactive_agents import ReactiveAgentBuilder, ReasoningStrategies

async def main():
    # Create an intelligent research agent
    agent = await (
        ReactiveAgentBuilder()
        .with_name("Research Assistant")
        .with_model("ollama:llama3")  # or "openai:gpt-4", "anthropic:claude-3-sonnet"
        .with_tools(["brave-search", "time"])  # Auto-detects MCP tools vs custom tools
        .with_instructions("Research thoroughly and provide detailed analysis")
        .with_reasoning_strategy(ReasoningStrategies.REACTIVE)
        .build()
    )

    async with agent:
        result = await agent.run(
            "What are the latest developments in quantum computing this week?"
        )
        print(result.final_answer)
        print(f"Status: {result.status_message}")

asyncio.run(main())

That's it! You now have a fully functional AI agent that can search the web, analyze information, and provide comprehensive answers.

🎯 Core Concepts

🤖 Agent Architecture

Reactive Agents uses a component-based architecture where each agent is composed of specialized, swappable components:

# The agent automatically manages these components:
ExecutionEngine  # Coordinates task execution and strategy selection
ReasoningEngine  # Handles different reasoning strategies
ToolManager     # Manages tool registration and execution
MemoryManager   # Handles persistent storage and retrieval
EventBus        # Coordinates real-time event communication
MetricsManager  # Tracks performance and provides insights

🧭 Reasoning Strategies

Choose the right strategy for your task:

from reactive_agents import ReactiveAgentBuilder, ReasoningStrategies

# Reactive: Fast, direct execution
agent = await ReactiveAgentBuilder().with_reasoning_strategy(ReasoningStrategies.REACTIVE).build()

# Plan-Execute-Reflect: Structured approach
agent = await ReactiveAgentBuilder().with_reasoning_strategy(ReasoningStrategies.PLAN_EXECUTE_REFLECT).build()

# Adaptive: AI selects the best strategy
agent = await ReactiveAgentBuilder().with_reasoning_strategy(ReasoningStrategies.ADAPTIVE).build()  # Default

🛠️ Tool Integration

Multiple ways to add capabilities to your agents:

from reactive_agents import tool

# 1. Custom Python functions with @tool decorator
@tool()
async def get_weather(city: str) -> str:
    """Get weather information for a city."""
    return f"Weather in {city}: Sunny, 72°F"

# 2. Mixed tools - auto-detection!
# Strings = MCP servers, Functions = custom tools
.with_tools([get_weather, "brave-search", "time", "filesystem"])

# 3. Or use explicit methods
.with_mcp_tools(["brave-search", "sqlite"])
.with_custom_tools([get_weather])

---

💡 Examples

🔍 Smart Research Agent

from reactive_agents import ReactiveAgentBuilder, tool, ReasoningStrategies

@tool()
async def analyze_trends(data: str) -> str:
    """Analyze data trends and patterns."""
    # Your analysis logic here
    return f"Trend analysis: {data}"

async def create_research_agent():
    return await (
        ReactiveAgentBuilder()
        .with_name("Research Pro")
        .with_model("openai:gpt-4")
        .with_reasoning_strategy(ReasoningStrategies.PLAN_EXECUTE_REFLECT)
        .with_tools([analyze_trends, "brave-search", "time", "filesystem"])
        .with_instructions("""
            You are a professional research analyst. Always:
            1. Search for the most recent information
            2. Cross-reference multiple sources
            3. Provide data-driven insights
            4. Save important findings to files
        """)
        .with_max_iterations(15)
        .build()
    )

📊 Business Intelligence Agent

async def create_bi_agent():
    return await (
        ReactiveAgentBuilder()
        .with_name("BI Analyst")
        .with_model("anthropic:claude-3-sonnet")
        .with_tools(["sqlite", "filesystem", "brave-search"])
        .with_vector_memory("bi_agent_memory")  # Enable persistent vector memory
        .with_instructions("""
            You are a business intelligence analyst. Create comprehensive
            reports with data visualizations and actionable insights.
        """)
        .with_response_format("""
            ## Executive Summary
            [Key findings and recommendations]

            ## Data Analysis
            [Detailed analysis with charts/tables]

            ## Recommendations
            [Specific, actionable next steps]
        """)
        .build()
    )

🔄 Multi-Agent Workflow

from reactive_agents.workflows import WorkflowOrchestrator

async def create_content_pipeline():
    orchestrator = WorkflowOrchestrator()

    # Research agent
    researcher = await (
        ReactiveAgentBuilder()
        .with_name("Content Researcher")
        .with_tools(["brave_web_search"])
        .build()
    )

    # Writing agent
    writer = await (
        ReactiveAgentBuilder()
        .with_name("Content Writer")
        .with_tools(["filesystem"])
        .build()
    )

    # Create workflow
    workflow = (
        orchestrator
        .add_agent("research", researcher)
        .add_agent("writing", writer)
        .add_dependency("writing", "research")  # Writer waits for researcher
        .build()
    )

    return workflow

🎛️ Event-Driven Monitoring

from reactive_agents.events import AgentStateEvent

async def create_monitored_agent():
    # Track performance in real-time
    metrics = {"tool_calls": 0, "errors": 0, "duration": 0}

    def on_tool_called(event):
        metrics["tool_calls"] += 1
        print(f"🔧 Tool used: {event['tool_name']}")

    def on_error(event):
        metrics["errors"] += 1
        print(f"❌ Error: {event['error_message']}")

    def on_completion(event):
        metrics["duration"] = event["total_duration"]
        print(f"✅ Completed in {metrics['duration']:.2f}s")
        print(f"📊 Final metrics: {metrics}")

    return await (
        ReactiveAgentBuilder()
        .with_name("Monitored Agent")
        .with_model("ollama:qwen2:7b")
        .on_tool_called(on_tool_called)
        .on_error_occurred(on_error)
        .on_session_ended(on_completion)
        .build()
    )

---

🛠️ Installation & Setup

Prerequisites

• Python 3.10+
• Poetry (recommended) or pip

Basic Installation

# Using pip
pip install reactive-agents

# Using Poetry
poetry add reactive-agents

Development Installation

# Clone the repository
git clone https://github.com/tylerjrbuell/reactive-agents
cd reactive-agents

# Install with Poetry
poetry install

# Run tests
poetry run pytest

Environment Configuration

Create a .env file:

# LLM Providers
OPENAI_API_KEY=your_openai_key
ANTHROPIC_API_KEY=your_anthropic_key
GROQ_API_KEY=your_groq_key
OLLAMA_HOST=http://localhost:11434

# MCP Tools
BRAVE_API_KEY=your_brave_search_key

# Optional: Custom MCP configuration
MCP_CONFIG_PATH=/path/to/custom/mcp_config.json

---

🎯 Advanced Features

🧠 Custom Reasoning Strategies

Implement your own reasoning approach:

from reactive_agents.strategies import BaseReasoningStrategy

class MyCustomStrategy(BaseReasoningStrategy):
    @property
    def name(self) -> str:
        return "my_custom_strategy"

    async def execute_iteration(self, task: str, context: ReasoningContext):
        # Your custom reasoning logic
        return StrategyResult.success(payload)

# Register and use
ReactiveAgentBuilder().with_reasoning_strategy("my_custom_strategy")

🔧 Custom Tool Creation

Build sophisticated tools with validation:

from reactive_agents.tools import tool
from pydantic import BaseModel

class WeatherRequest(BaseModel):
    city: str
    units: str = "metric"

@tool("Get detailed weather information", validation_model=WeatherRequest)
async def advanced_weather(request: WeatherRequest) -> dict:
    # Sophisticated weather logic with API calls
    weather_data = await fetch_weather_api(request.city, request.units)
    return {
        "temperature": weather_data.temp,
        "conditions": weather_data.conditions,
        "forecast": weather_data.forecast
    }

📊 Performance Monitoring

Track and optimize agent performance:

async def monitor_performance():
    agent = await ReactiveAgentBuilder().with_name("Performance Agent").build()

    # Get real-time metrics
    session = agent.context.session

    print(f"Completion Score: {session.completion_score}")
    print(f"Tool Usage Score: {session.tool_usage_score}")
    print(f"Overall Score: {session.overall_score}")

    # Access detailed metrics
    metrics = agent.context.metrics_manager.get_metrics()
    print(f"Total Duration: {metrics['total_time']:.2f}s")
    print(f"Tool Calls: {metrics['tool_calls']}")
    print(f"Model Calls: {metrics['model_calls']}")

🔄 Plugin System 🚧

Extend the framework with plugins:

from reactive_agents.plugins import Plugin

class CustomAnalyticsPlugin(Plugin):
    def on_load(self, framework):
        # Initialize your plugin
        self.analytics_client = AnalyticsClient()

    def on_agent_created(self, agent):
        # Hook into agent lifecycle
        agent.on_completion(self.track_completion)

    async def track_completion(self, event):
        await self.analytics_client.track(event)

# Load plugin
framework.load_plugin(CustomAnalyticsPlugin())

---

📖 Documentation

📚 Comprehensive Guides

• [Getting Started Guide](docs/getting-started.md) - Your first agent in 5 minutes
• [Architecture Overview](docs/architecture.md) - Understanding the framework
• [Tool Development](docs/tools.md) - Building custom tools and integrations
• [Reasoning Strategies](docs/strategies.md) - Deep dive into AI reasoning
• [Workflow Orchestration](docs/workflows.md) - Multi-agent coordination
• [Production Deployment](docs/deployment.md) - Scaling to production

🔧 API Reference

• [Agent Builder API](docs/api/builder.md) - Complete builder pattern reference
• [Tool System API](docs/api/tools.md) - Tool registration and execution
• [Event System API](docs/api/events.md) - Real-time monitoring and hooks
• [Configuration API](docs/api/config.md) - Advanced configuration options

💡 Examples & Tutorials

• [Example Gallery](examples/) - 20+ real-world examples
• [Tutorial Series](docs/tutorials/) - Step-by-step learning path
• [Best Practices](docs/best-practices.md) - Production tips and patterns
• [Troubleshooting](docs/troubleshooting.md) - Common issues and solutions

---

🌐 Model Provider Support

Reactive Agents works with all major LLM providers:

| Provider | Models | Features | | ------------- | --------------------------- | ----------------------------------- | | OpenAI | GPT-4o, GPT-4, GPT-3.5 | Function calling, streaming, vision | | Anthropic | Claude 3.5 Sonnet, Claude 3 | Large context, tool use | | Groq | Llama 3, Mixtral | Ultra-fast inference | | Ollama | Any local model | Privacy,

…

Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: tylerjrbuell
• Source: tylerjrbuell/reactive-agents
• License: MIT

Install and usage instructions live in the source repository linked above.