""" Flask API server for InklyAI AgentAI integration. """ from flask import Flask, request, jsonify from flask_cors import CORS import logging import os from datetime import datetime import json from agentai_integration import AgentAISignatureManager, AgentAISignatureAPI # Initialize Flask app app = Flask(__name__) CORS(app) # Enable CORS for cross-origin requests # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Initialize signature manager signature_manager = AgentAISignatureManager( model_path=None, # Use default model threshold=0.75, device='auto' ) # Initialize API wrapper api = AgentAISignatureAPI(signature_manager) @app.route('/health', methods=['GET']) def health_check(): """Health check endpoint.""" return jsonify({ 'status': 'healthy', 'timestamp': datetime.now().isoformat(), 'service': 'InklyAI AgentAI Integration' }) @app.route('/register-agent', methods=['POST']) def register_agent(): """Register a new agent with signature template.""" try: data = request.get_json() if not data or 'agent_id' not in data or 'signature_template' not in data: return jsonify({ 'success': False, 'error': 'Missing required fields: agent_id, signature_template' }), 400 result = api.register_agent_endpoint(data) if result['success']: return jsonify(result), 200 else: return jsonify(result), 400 except Exception as e: logger.error(f"Registration error: {e}") return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/verify-signature', methods=['POST']) def verify_signature(): """Verify agent signature.""" try: data = request.get_json() if not data or 'agent_id' not in data or 'signature_image' not in data: return jsonify({ 'success': False, 'error': 'Missing required fields: agent_id, signature_image' }), 400 result = api.verify_signature_endpoint(data) if result['success']: return jsonify(result), 200 else: return jsonify(result), 400 except Exception as e: logger.error(f"Verification error: {e}") return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/batch-verify', methods=['POST']) def batch_verify(): """Batch verify multiple agent signatures.""" try: data = request.get_json() if not data or 'verification_requests' not in data: return jsonify({ 'success': False, 'error': 'Missing required field: verification_requests' }), 400 verification_requests = data['verification_requests'] results = signature_manager.batch_verify_agents(verification_requests) # Convert results to serializable format serializable_results = [] for result in results: serializable_results.append({ 'verification_id': result.verification_id, 'agent_id': result.agent_id, 'is_verified': result.is_verified, 'similarity_score': result.similarity_score, 'confidence': result.confidence, 'timestamp': result.timestamp.isoformat() }) return jsonify({ 'success': True, 'results': serializable_results, 'total_processed': len(serializable_results) }), 200 except Exception as e: logger.error(f"Batch verification error: {e}") return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/agent-stats/', methods=['GET']) def get_agent_stats(agent_id): """Get verification statistics for an agent.""" try: result = api.get_stats_endpoint(agent_id) if result['success']: return jsonify(result), 200 else: return jsonify(result), 400 except Exception as e: logger.error(f"Stats error: {e}") return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/deactivate-agent/', methods=['POST']) def deactivate_agent(agent_id): """Deactivate an agent.""" try: success = signature_manager.deactivate_agent(agent_id) return jsonify({ 'success': success, 'agent_id': agent_id, 'action': 'deactivated', 'timestamp': datetime.now().isoformat() }), 200 if success else 404 except Exception as e: logger.error(f"Deactivation error: {e}") return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/reactivate-agent/', methods=['POST']) def reactivate_agent(agent_id): """Reactivate an agent.""" try: success = signature_manager.reactivate_agent(agent_id) return jsonify({ 'success': success, 'agent_id': agent_id, 'action': 'reactivated', 'timestamp': datetime.now().isoformat() }), 200 if success else 404 except Exception as e: logger.error(f"Reactivation error: {e}") return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/list-agents', methods=['GET']) def list_agents(): """List all registered agents.""" try: agents = [] for agent_id, agent_signature in signature_manager.agent_signatures.items(): agents.append({ 'agent_id': agent_id, 'created_at': agent_signature.created_at.isoformat(), 'last_verified': agent_signature.last_verified.isoformat() if agent_signature.last_verified else None, 'verification_count': agent_signature.verification_count, 'is_active': agent_signature.is_active }) return jsonify({ 'success': True, 'agents': agents, 'total_agents': len(agents) }), 200 except Exception as e: logger.error(f"List agents error: {e}") return jsonify({ 'success': False, 'error': str(e) }), 500 @app.route('/config', methods=['GET']) def get_config(): """Get current configuration.""" return jsonify({ 'success': True, 'config': signature_manager.config, 'threshold': signature_manager.verifier.threshold, 'device': str(signature_manager.verifier.device) }), 200 @app.route('/config', methods=['POST']) def update_config(): """Update configuration.""" try: data = request.get_json() if 'threshold' in data: signature_manager.verifier.threshold = data['threshold'] if 'config' in data: signature_manager.config.update(data['config']) return jsonify({ 'success': True, 'message': 'Configuration updated', 'timestamp': datetime.now().isoformat() }), 200 except Exception as e: logger.error(f"Config update error: {e}") return jsonify({ 'success': False, 'error': str(e) }), 500 @app.errorhandler(404) def not_found(error): """Handle 404 errors.""" return jsonify({ 'success': False, 'error': 'Endpoint not found' }), 404 @app.errorhandler(500) def internal_error(error): """Handle 500 errors.""" return jsonify({ 'success': False, 'error': 'Internal server error' }), 500 if __name__ == '__main__': # Register some example agents for testing try: # Register sample agents if sample data exists if os.path.exists('data/samples/john_doe_1.png'): signature_manager.register_agent_signature( 'demo_agent_001', 'data/samples/john_doe_1.png' ) logger.info("Registered demo agent 001") if os.path.exists('data/samples/jane_smith_1.png'): signature_manager.register_agent_signature( 'demo_agent_002', 'data/samples/jane_smith_1.png' ) logger.info("Registered demo agent 002") logger.info("Demo agents registered successfully") except Exception as e: logger.warning(f"Could not register demo agents: {e}") # Start the Flask server port = int(os.environ.get('PORT', 5000)) debug = os.environ.get('DEBUG', 'False').lower() == 'true' logger.info(f"Starting InklyAI AgentAI Integration API on port {port}") app.run(host='0.0.0.0', port=port, debug=debug)