Jack demo: MQ Server handshake test (retry 1)

{
  "cost": {
    "input_tokens": 63,
    "estimated_usd": 0.071739,
    "output_tokens": 4770
  },
  "model": "claude-sonnet-4-20250514",
  "status": "success",
  "transport": "anthropic_api",
  "latency_ms": 63405,
  "response_text": "I'll help you create an MQ server handshake test for the CBP ABI system with ISA/GS envelope generation for Entry Summary. Let me create a test script based on the demo_jack_7501.py reference pattern.\n\n```python\n#!/usr/bin/env python3\n\"\"\"\nJack Demo: MQ Server Handshake Test (Retry 1)\nCBP ABI Entry Summary (Type 01) with ISA/GS Envelope Generation\n\"\"\"\n\nimport sys\nimport json\nimport time\nimport logging\nfrom datetime import datetime\nfrom typing import Dict, Any, Optional, Tuple\nimport pymqi\nimport uuid\n\n# Configure logging\nlogging.basicConfig(\n    level=logging.INFO,\n    format='%(asctime)s - %(levelname)s - %(message)s',\n    handlers=[\n        logging.FileHandler('jack_demo_mq_test.log'),\n        logging.StreamHandler(sys.stdout)\n    ]\n)\nlogger = logging.getLogger(__name__)\n\nclass CBPABIClient:\n    \"\"\"CBP ABI MQ Client for Entry Summary processing\"\"\"\n    \n    def __init__(self, config: Dict[str, Any]):\n        self.config = config\n        self.qmgr = None\n        self.queue = None\n        self.reply_queue = None\n        self.connected = False\n        \n    def connect(self) -> bool:\n        \"\"\"Establish MQ connection to CBP ABI system\"\"\"\n        try:\n            logger.info(\"Initiating MQ connection to CBP ABI system...\")\n            \n            # Connection parameters\n            conn_info = f\"{self.config['host']}({self.config['port']})\"\n            \n            # Connect to queue manager\n            cd = pymqi.CD()\n            cd.ChannelName = self.config['channel']\n            cd.ConnectionName = conn_info\n            cd.ChannelType = pymqi.CMQC.MQCHT_CLNTCONN\n            cd.TransportType = pymqi.CMQC.MQXPT_TCP\n            \n            # SSL/TLS configuration if required\n            if self.config.get('ssl_enabled', True):\n                cd.SSLCipherSpec = self.config.get('ssl_cipher', 'TLS_RSA_WITH_AES_256_CBC_SHA256')\n                \n            self.qmgr = pymqi.connect(\n                self.config['queue_manager'], \n                cd=cd,\n                user=self.config.get('username', ''),\n                password=self.config.get('password', '')\n            )\n            \n            # Open request queue\n            self.queue = pymqi.Queue(self.qmgr, self.config['request_queue'])\n            \n            # Open reply queue\n            self.reply_queue = pymqi.Queue(self.qmgr, self.config['reply_queue'])\n            \n            self.connected = True\n            logger.info(\"✓ MQ connection established successfully\")\n            return True\n            \n        except Exception as e:\n            logger.error(f\"✗ MQ connection failed: {str(e)}\")\n            return False\n    \n    def generate_control_numbers(self) -> Tuple[str, str, str]:\n        \"\"\"Generate control numbers for ISA, GS, and ST segments\"\"\"\n        timestamp = datetime.now().strftime(\"%Y%m%d%H%M%S\")\n        \n        isa_control = f\"ISA{timestamp[-9:]}\"[:9]  # ISA control number (9 digits)\n        gs_control = f\"GS{timestamp[-9:]}\"[:9]    # GS control number (9 digits) \n        st_control = f\"ST{timestamp[-4:]}\"[:4]    # ST control number (4 digits)\n        \n        return isa_control, gs_control, st_control\n    \n    def generate_isa_envelope(self, control_num: str) -> str:\n        \"\"\"Generate ISA (Interchange Control Header) segment\"\"\"\n        now = datetime.now()\n        \n        isa_segments = [\n            \"ISA\",\n            \"00\",  # Authorization Information Qualifier\n            \" \" * 10,  # Authorization Information\n            \"00\",  # Security Information Qualifier  \n            \" \" * 10,  # Security Information\n            \"ZZ\",  # Interchange ID Qualifier (Sender)\n            self.config['sender_id'].ljust(15),  # Interchange Sender ID\n            \"ZZ\",  # Interchange ID Qualifier (Receiver)\n            self.config['receiver_id'].ljust(15),  # Interchange Receiver ID\n            now.strftime(\"%y%m%d\"),  # Interchange Date (YYMMDD)\n            now.strftime(\"%H%M\"),  # Interchange Time (HHMM)\n            \"U\",  # Interchange Control Standards ID\n            \"00401\",  # Interchange Control Version Number\n            control_num.zfill(9),  # Interchange Control Number\n            \"0\",  # Acknowledgment Requested\n            self.config.get('test_indicator', 'T'),  # Usage Indicator (T=Test, P=Production)\n            \">\"  # Component Element Separator\n        ]\n        \n        return \"*\".join(isa_segments) + \"~\"\n    \n    def generate_gs_envelope(self, control_num: str) -> str:\n        \"\"\"Generate GS (Functional Group Header) segment\"\"\"\n        now = datetime.now()\n        \n        gs_segments = [\n            \"GS\",\n            \"QM\",  # Functional Identifier Code (QM = Response)\n            self.config['sender_id'],  # Application Sender Code\n            self.config['receiver_id'],  # Application Receiver Code  \n            now.strftime(\"%Y%m%d\"),  # Date (YYYYMMDD)\n            now.strftime(\"%H%M%S\"),  # Time (HHMMSS)\n            control_num.zfill(9),  # Group Control Number\n            \"X\",  # Responsible Agency Code\n            \"004010\"  # Version/Release/Industry ID\n        ]\n        \n        return \"*\".join(gs_segments) + \"~\"\n    \n    def generate_entry_summary_01(self, st_control: str) -> str:\n        \"\"\"Generate Entry Summary (Type 01) transaction\"\"\"\n        \n        # ST - Transaction Set Header\n        st_segment = f\"ST*214*{st_control.zfill(4)}~\"\n        \n        # Entry Summary Type 01 segments\n        entry_data = [\n            # QM - Transportation Reference Number\n            f\"QM*01*{self.config.get('entry_number', 'TEST12345')}~\",\n            \n            # N1 - Name segments for parties\n            f\"N1*SF*{self.config.get('importer_name', 'TEST IMPORTER')}*91*{self.config.get('importer_id', '123456789')}~\",\n            \n            # N4 - Geographic Location  \n            f\"N4*{self.config.get('port_code', 'USNYC')}*NY*10001*US~\",\n            \n            # DTM - Date/Time segments\n            f\"DTM*102*{datetime.now().strftime('%Y%m%d')}~\",  # Estimated arrival\n            \n            # Test entry summary line items\n            \"LX*1~\",  # Assigned Number\n            f\"N1*VN*TEST VENDOR*91*TESTVENDOR001~\",\n            \"N4*TEST CITY*NY*10001*US~\",\n            \n            # SE - Transaction Set Trailer\n            f\"SE*8*{st_control.zfill(4)}~\"\n        ]\n        \n        return st_segment + \"\".join(entry_data)\n    \n    def generate_ge_trailer(self, control_num: str, segment_count: int = 1) -> str:\n        \"\"\"Generate GE (Functional Group Trailer) segment\"\"\"\n        return f\"GE*{segment_count}*{control_num.zfill(9)}~\"\n    \n    def generate_iea_trailer(self, control_num: str, group_count: int = 1) -> str:\n        \"\"\"Generate IEA (Interchange Control Trailer) segment\"\"\"\n        return f\"IEA*{group_count}*{control_num.zfill(9)}~\"\n    \n    def create_edi_message(self) -> str:\n        \"\"\"Create complete EDI message with ISA/GS envelope\"\"\"\n        logger.info(\"Generating EDI Entry Summary (Type 01) message...\")\n        \n        # Generate control numbers\n        isa_control, gs_control, st_control = self.generate_control_numbers()\n        \n        logger.info(f\"Control Numbers - ISA: {isa_control}, GS: {gs_control}, ST: {st_control}\")\n        \n        # Build EDI message\n        edi_parts = [\n            self.generate_isa_envelope(isa_control),\n            self.generate_gs_envelope(gs_control), \n            self.generate_entry_summary_01(st_control),\n            self.generate_ge_trailer(gs_control),\n            self.generate_iea_trailer(isa_control)\n        ]\n        \n        edi_message = \"\".join(edi_parts)\n        \n        logger.info(f\"✓ EDI message generated ({len(edi_message)} characters)\")\n        return edi_message\n    \n    def send_entry_summary(self) -> Optional[str]:\n        \"\"\"Send Entry Summary to CBP ABI system\"\"\"\n        if not self.connected:\n            logger.error(\"Not connected to MQ server\")\n            return None\n            \n        try:\n            # Create EDI message\n            edi_message = self.create_edi_message()\n            \n            # Create message descriptor\n            md = pymqi.MD()\n            md.Format = pymqi.CMQC.MQFMT_STRING\n            md.MessageType = pymqi.CMQC.MQMT_REQUEST\n            md.ReplyToQ = self.config['reply_queue']\n            md.MessageId = pymqi.CMQC.MQMI_NONE\n            md.CorrelId = str(uuid.uuid4()).encode()[:24]\n            \n            # Put message options\n            pmo = pymqi.PMO()\n            pmo.Options = pymqi.CMQC.MQPMO_NEW_MSG_ID\n            \n            logger.info(\"Sending Entry Summary to CBP ABI system...\")\n            \n            # Send message\n            self.queue.put(edi_message.encode('utf-8'), md, pmo)\n            \n            logger.info(f\"✓ Message sent successfully. CorrelID: {md.CorrelId}\")\n            \n            return md.CorrelId\n            \n        except Exception as e:\n            logger.error(f\"✗ Failed to send message: {str(e)}\")\n            return None\n    \n    def wait_for_response(self, correlation_id: bytes, timeout: int = 30) -> Optional[str]:\n        \"\"\"Wait for response from CBP ABI system\"\"\"\n        try:\n            logger.info(f\"Waiting for response (timeout: {timeout}s)...\")\n            \n            # Message descriptor for getting response\n            md = pymqi.MD()\n            md.CorrelId = correlation_id\n            \n            # Get message options  \n            gmo = pymqi.GMO()\n            gmo.Options = pymqi.CMQC.MQGMO_WAIT | pymqi.CMQC.MQGMO_MATCH_CORREL_ID\n            gmo.WaitInterval = timeout * 1000  # milliseconds\n            \n            # Get response message\n            response_data = self.reply_queue.get(None, md, gmo)\n            \n            response_text = response_data.decode('utf-8')\n            logger.info(f\"✓ Response received ({len(response_text)} characters)\")\n            \n            return response_text\n            \n        except pymqi.MQMIError as e:\n            if e.reason == pymqi.CMQC.MQRC_NO_MSG_AVAILABLE:\n                logger.warning(\"✗ No response received within timeout period\")\n            else:\n                logger.error(f\"✗ Error receiving response: {e}\")\n            return None\n        except Exception as e:\n            logger.error(f\"✗ Unexpected error: {str(e)}\")\n            return None\n    \n    def disconnect(self):\n        \"\"\"Close MQ connections\"\"\"\n        try:\n            if self.queue:\n                self.queue.close()\n            if self.reply_queue:\n                self.reply_queue.close()\n            if self.qmgr:\n                self.qmgr.disconnect()\n                \n            self.connected = False\n            logger.info(\"✓ MQ connection closed\")\n            \n        except Exception as e:\n            logger.error(f\"Error during disconnect: {str(e)}\")\n\ndef load_config() -> Dict[str, Any]:\n    \"\"\"Load CBP ABI configuration\"\"\"\n    # Demo configuration - replace with actual CBP ABI settings\n    config = {\n        # MQ Connection settings\n        'host': 'cbp-abi-test.dhs.gov',  # Replace with actual CBP host\n        'port': 1414,\n        'channel': 'CBP.ABI.SVRCONN',\n        'queue_manager': 'CBP.ABI.QM',\n        'request_queue': 'CBP.ABI.ENTRY.REQUEST',\n        'reply_queue': 'CBP.ABI.ENTRY.RESPONSE',\n        \n        # Authentication\n        'username': 'ABI_USER',  # Your CBP ABI username\n        'password': 'ABI_PASS',  # Your CBP ABI password  \n        \n        # SSL/Security\n        'ssl_enabled': True,\n        'ssl_cipher': 'TLS_RSA_WITH_AES_256_CBC_SHA256',\n        \n        # EDI Identifiers\n        'sender_id': 'TESTCLIENT01',      # Your assigned sender ID\n        'receiver_id': 'CBPABI',          # CBP ABI receiver ID\n        'test_indicator': 'T',            # T=Test, P=Production\n        \n        # Entry Summary data\n        'entry_number': 'TEST20240101001',\n        'importer_name': 'DEMO IMPORT COMPANY',\n        'importer_id': '1234567890',\n        'port_code': 'USNYC'  # Port of arrival\n    }\n    \n    return config\n\ndef run_handshake_test():\n    \"\"\"Execute CBP ABI MQ handshake test\"\"\"\n    logger.info(\"=== Jack Demo: CBP ABI MQ Handshake Test (Retry 1) ===\")\n    \n    # Load configuration\n    config = load_config()\n    \n    # Initialize client\n    client = CBPABIClient(config)\n    \n    try:\n        # Test connection\n        if not client.connect():\n            logger.error(\"❌ Handshake test failed - could not connect\")\n            return False\n        \n        # Send Entry Summary\n        correlation_id = client.send_entry_summary()\n        if not correlation_id:\n            logger.error(\"❌ Handshake test failed - could not send message\")\n            return False\n        \n        # Wait for response\n        response = client.wait_for_response(correlation_id, timeout=30)\n        \n        if response:\n            logger.info(\"✅ Handshake test successful!\")\n            logger.info(\"Response preview:\")\n            logger.info(response[:200] + \"...\" if len(response) > 200 else response)\n            \n            # Save full response\n            with open(f\"cbp_response_{int(time.time())}.edi\", \"w\") as f:\n                f.write(response)\n            logger.info(\"Full response saved to file\")\n            \n            return True\n        else:\n            logger.warning(\"⚠️ Handshake partially successful - no response received\")\n            return False\n            \n    except Exception as e:\n        logger.error(f\"❌ Handshake test failed with exception: {str(e)}\")\n        return False\n        \n    finally:\n        client.disconnect()\n\ndef main():\n    \"\"\"Main execution function\"\"\"\n    try:\n        success = run_handshake_test()\n        \n        if success:\n            logger.info(\"\\n🎉 CBP ABI MQ handshake test completed successfully!\")\n            sys.exit(0)\n        else:\n            logger.error(\"\\n💥 CBP ABI MQ handshake test failed!\")\n            sys.exit(1)\n            \n    except KeyboardInterrupt:\n        logger.info(\"\\n⏹️ Test interrupted by user\")\n        sys.exit(1)\n    except Exception as e:\n        logger.error(f\"\\n💥 Unexpected error: {str(e)}\")\n        sys.exit(1)\n\nif __name__ == \"__main__\":\n    main()\n```\n\nThis script also needs a requirements file:\n\n```txt\n# requirements.txt\npymqi>=1.12.0\n```\n\nAnd here's a configuration template file:\n\n```json\n{\n  \"cbp_abi_config_template.json\": {\n    \"mq_connection\": {\n      \"host\": \"cbp-abi-test.dhs.gov\",\n      \"port\": 1414,\n      \"channel\": \"CBP.ABI.SVRCONN\", \n      \"queue_manager\": \"CBP.ABI.QM\",\n      \"request_queue\": \"CBP.ABI.ENTRY.REQUEST\",\n      \"reply_queue\": \"CBP.ABI.ENTRY.RESPONSE\"\n    },\n    \"authentication\": {\n      \"username\": \"YOUR_ABI_USERNAME\",\n      \"password\": \"YOUR_ABI_PASSWORD\"\n    },\n    \"edi_settings\": {\n      \"sender_id\": \"YOUR_SENDER_ID\",\n      \"receiver_id\": \"CBPABI\",\n      \"test_indicator\": \"T\"\n    },\n    \"entry_summary\": {\n      \"entry_number\": \"TEST20240101001\",\n      \"importer_name\": \"YOUR COMPANY NAME\",\n      \"importer_id\": \"YOUR_IMPORTER_ID\",\n      \"port_code\": \"USNYC\"\n    }\n  }\n}\n```\n\n## Key Features:\n\n1. **MQ Connection Management**: Handles IBM MQ connection with SSL/TLS support\n2. **ISA/GS Envelope Generation**: Creates proper EDI X12 envelope structure\n3. **Entry Summary Type 01**: Generates CBP ABI Entry Summary transaction\n4. **Control Number Generation**: Creates unique ISA, GS, and ST control numbers\n5. **Error Handling**: Comprehensive error handling and logging\n6. **Response Processing**: Waits for and processes CBP ABI responses\n\n## Usage:\n\n```bash\n# Install dependencies\npip install -r requirements.txt\n\n# Run the test\npython jack_demo_mq_handshake_test.py\n```\n\n**Important Notes:**\n- Replace configuration values with your actual CBP ABI credentials\n- Ensure you have proper CBP ABI system access and certificates\n- Test in CBP's test environment before production use\n- Follow CBP ABI documentation for exact message formats\n\nWould you like me to modify any specific aspects of the handshake test or add additional functionality?"
}