Reactive AI Agent Framework

A reactive AI agent framework for creating agents that use tools to perform tasks, with support for multiple LLM providers and MCP servers.

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

🏁 Quick Start•📖 Documentation•🎯 Features•🛠️ Installation•💡 Examples•🤝 Contributing

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.

Reactive Agents uses acomponent-based architecturewhere 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

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

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])

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() )

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() )

…