A flexible, extensible PocketFlow agent that converts natural language queries into database queries with support for multiple database types, intelligent schema analysis, and seamless integration into other agentic systems.
- Multi-Database Support: PostgreSQL, MySQL, SQLite, SQL Server, and more
- Intelligent Query Analysis: Understands query intent, complexity, and requirements
- Smart Schema Management: Context-aware schema extraction and caching
- Advanced Error Handling: Multi-stage validation and correction
- Security-First: SQL injection prevention and access control
- Performance Optimized: Connection pooling and query caching
- Plugin Architecture: Extensible components for custom database types
- API Interface: RESTful API for external system integration
- Event System: Callbacks and hooks for monitoring and customization
- Configuration Management: Environment-based and programmatic configuration
- Multiple Output Formats: JSON, CSV, formatted tables, and custom formats
- Query Classification: Automatically categorizes queries (SELECT, INSERT, analytics, etc.)
- Result Visualization: Built-in charting and data visualization options
- Query History & Caching: Intelligent caching of similar queries
- Access Control: Role-based permissions and query restrictions
- Monitoring & Logging: Comprehensive observability features
The agent uses a modular, plugin-based architecture with these core components:
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ Query Input │ -> │ Query Analyzer │ -> │ Schema Manager │
└─────────────────┘ └──────────────────┘ └─────────────────┘
│
▼
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ Result Formatter│ <- │ Query Executor │ <- │ Query Generator │
└─────────────────┘ └──────────────────┘ └─────────────────┘
│
▼
┌──────────────────┐
│ Error Handler │
└──────────────────┘
# Clone or create the project
cd cookbook/pocketflow-db-agent
# Install dependencies
pip install -r requirements.txt
# Set up configuration
cp .env.example .env
# Edit .env with your settingsfrom pocketflow_db_agent import DatabaseAgent
# Initialize agent
agent = DatabaseAgent(
database_url="postgresql://user:pass@localhost/mydb",
# or use configuration file
# config_file="config.yaml"
)
# Simple query
result = agent.query("Show me all customers from New York")
print(result.data) # Formatted results
print(result.sql) # Generated SQL
print(result.metadata) # Execution metadata
# Advanced usage with options
result = agent.query(
"What are the top 5 products by revenue this month?",
options={
"format": "json",
"include_metadata": True,
"cache": True,
"validate": True
}
)# As a PocketFlow node
from pocketflow_db_agent.nodes import DatabaseQueryNode
db_node = DatabaseQueryNode(
database_url="sqlite:///data.db",
config={"max_retries": 3}
)
# Use in a flow
query_flow = Flow(start=db_node)
result = query_flow.run({"natural_query": "Count all users"})# Start the API server
python -m pocketflow_db_agent.server --port 8080
# Use the REST API
curl -X POST http://localhost:8080/query \
-H "Content-Type: application/json" \
-d '{"query": "List all active users", "database": "main"}'- Fork the repository
- Create a feature branch
- Make your changes
- Add tests
- Submit a pull request
This project is licensed under the MIT License - see the LICENSE file for details.