| """ |
| 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 |
|
|
| |
| app = Flask(__name__) |
| CORS(app) |
|
|
| |
| logging.basicConfig(level=logging.INFO) |
| logger = logging.getLogger(__name__) |
|
|
| |
| signature_manager = AgentAISignatureManager( |
| model_path=None, |
| threshold=0.75, |
| device='auto' |
| ) |
|
|
| |
| 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) |
| |
| |
| 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/<agent_id>', 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/<agent_id>', 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/<agent_id>', 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__': |
| |
| try: |
| |
| 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}") |
| |
| |
| 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) |
|
|
