flaskapi

Flask API Skill

Provides comprehensive Flask API development capabilities for the Golden Armada AI Agent Fleet Platform.

When to Use This Skill

Activate this skill when working with:

• Flask application development
• Blueprint organization
• Flask extensions
• REST API patterns
• Flask deployment

Application Structure

``` app/ ├── init.py # Application factory ├── config.py # Configuration ├── extensions.py # Flask extensions ├── models/ │ ├── init.py │ └── agent.py ├── api/ │ ├── init.py │ ├── agents.py # Agents blueprint │ └── tasks.py # Tasks blueprint ├── services/ │ ├── init.py │ └── agent_service.py └── utils/ └── validators.py ```

Application Factory

```python

app/init.py

from flask import Flask from app.config import Config from app.extensions import db, migrate, cors

def create_app(config_class=Config): app = Flask(name) app.config.from_object(config_class)

# Initialize extensions
db.init_app(app)
migrate.init_app(app, db)
cors.init_app(app)

# Register blueprints
from app.api.agents import agents_bp
from app.api.tasks import tasks_bp

app.register_blueprint(agents_bp, url_prefix='/api/v1/agents')
app.register_blueprint(tasks_bp, url_prefix='/api/v1/tasks')

# Register error handlers
register_error_handlers(app)

# Health check
@app.route('/health')
def health():
    return {'status': 'healthy', 'service': 'golden-armada'}

return app

def register_error_handlers(app): @app.errorhandler(404) def not_found(error): return {'error': 'Not found'}, 404

@app.errorhandler(500)
def internal_error(error):
    return {'error': 'Internal server error'}, 500

```

Configuration

```python

app/config.py

import os from datetime import timedelta

class Config: SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-secret-key') SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL', 'sqlite:///app.db') SQLALCHEMY_TRACK_MODIFICATIONS = False JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY', 'jwt-secret') JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)

class DevelopmentConfig(Config): DEBUG = True

class ProductionConfig(Config): DEBUG = False

class TestingConfig(Config): TESTING = True SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:' ```

Extensions

```python

app/extensions.py

from flask_sqlalchemy import SQLAlchemy from flask_migrate import Migrate from flask_cors import CORS from flask_jwt_extended import JWTManager

db = SQLAlchemy() migrate = Migrate() cors = CORS() jwt = JWTManager() ```

Models

```python

app/models/agent.py

from app.extensions import db from datetime import datetime import uuid

class Agent(db.Model): tablename = 'agents'

id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
name = db.Column(db.String(100), nullable=False)
type = db.Column(db.String(50), nullable=False)
status = db.Column(db.String(20), default='idle')
config = db.Column(db.JSON, default={})
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

tasks = db.relationship('Task', backref='agent', lazy='dynamic', cascade='all, delete-orphan')

def to_dict(self):
    return {
        'id': self.id,
        'name': self.name,
        'type': self.type,
        'status': self.status,
        'config': self.config,
        'created_at': self.created_at.isoformat(),
        'updated_at': self.updated_at.isoformat()
    }

```

Blueprints (API Routes)

```python

app/api/agents.py

from flask import Blueprint, request, jsonify from app.models.agent import Agent from app.extensions import db from app.utils.validators import validate_agent_input

agents_bp = Blueprint('agents', name)

@agents_bp.route('/', methods=['GET']) def list_agents(): """List all agents with optional filtering.""" agent_type = request.args.get('type') status = request.args.get('status') page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int)

query = Agent.query

if agent_type:
    query = query.filter(Agent.type == agent_type)
if status:
    query = query.filter(Agent.status == status)

pagination = query.paginate(page=page, per_page=per_page)

return jsonify({
    'agents': [agent.to_dict() for agent in pagination.items],
    'total': pagination.total,
    'pages': pagination.pages,
    'current_page': page
})

@agents_bp.route('/<agent_id>', methods=['GET']) def get_agent(agent_id): """Get a specific agent by ID.""" agent = Agent.query.get_or_404(agent_id) return jsonify(agent.to_dict())

@agents_bp.route('/', methods=['POST']) def create_agent(): """Create a new agent.""" data = request.get_json()

errors = validate_agent_input(data)
if errors:
    return jsonify({'errors': errors}), 400

agent = Agent(
    name=data['name'],
    type=data['type'],
    config=data.get('config', {})
)

db.session.add(agent)
db.session.commit()

return jsonify(agent.to_dict()), 201

@agents_bp.route('/<agent_id>', methods=['PUT']) def update_agent(agent_id): """Update an existing agent.""" agent = Agent.query.get_or_404(agent_id) data = request.get_json()

if 'name' in data:
    agent.name = data['name']
if 'config' in data:
    agent.config = data['config']
if 'status' in data:
    agent.status = data['status']

db.session.commit()
return jsonify(agent.to_dict())

@agents_bp.route('/<agent_id>', methods=['DELETE']) def delete_agent(agent_id): """Delete an agent.""" agent = Agent.query.get_or_404(agent_id) db.session.delete(agent) db.session.commit() return '', 204 ```

JWT Authentication

```python

app/api/auth.py

from flask import Blueprint, request, jsonify from flask_jwt_extended import ( create_access_token, create_refresh_token, jwt_required, get_jwt_identity ) from app.models.user import User from app.extensions import db

auth_bp = Blueprint('auth', name)

@auth_bp.route('/login', methods=['POST']) def login(): data = request.get_json() user = User.query.filter_by(email=data.get('email')).first()

if user and user.check_password(data.get('password')):
    access_token = create_access_token(identity=user.id)
    refresh_token = create_refresh_token(identity=user.id)
    return jsonify({
        'access_token': access_token,
        'refresh_token': refresh_token
    })

return jsonify({'error': 'Invalid credentials'}), 401

@auth_bp.route('/refresh', methods=['POST']) @jwt_required(refresh=True) def refresh(): identity = get_jwt_identity() access_token = create_access_token(identity=identity) return jsonify({'access_token': access_token})

@auth_bp.route('/me', methods=['GET']) @jwt_required() def get_current_user(): user_id = get_jwt_identity() user = User.query.get(user_id) return jsonify(user.to_dict()) ```

Request Validation

```python

app/utils/validators.py

def validate_agent_input(data): errors = []

if not data.get('name'):
    errors.append('Name is required')
elif len(data['name']) > 100:
    errors.append('Name must be less than 100 characters')

valid_types = ['claude', 'gpt', 'gemini', 'ollama']
if not data.get('type'):
    errors.append('Type is required')
elif data['type'] not in valid_types:
    errors.append(f'Type must be one of: {", ".join(valid_types)}')

return errors

```

CLI Commands

```python

app/cli.py

import click from flask.cli import with_appcontext from app.extensions import db

@click.command('init-db') @with_appcontext def init_db_command(): """Initialize the database.""" db.create_all() click.echo('Initialized the database.')

@click.command('seed-db') @with_appcontext def seed_db_command(): """Seed the database with sample data.""" from app.models.agent import Agent agents = [ Agent(name='Claude Agent', type='claude'), Agent(name='GPT Agent', type='gpt'), ] db.session.add_all(agents) db.session.commit() click.echo('Seeded the database.') ```

Run Commands

```bash

Development

flask run --debug

Production (Gunicorn)

gunicorn -w 4 -b 0.0.0.0:8000 "app:create_app()"

Database migrations

flask db init flask db migrate -m "Initial migration" flask db upgrade ```