Repo restructuring for better readability.

agentic-healthcare-booking-app/common/identity/tbac.py

@@ -0,0 +1,138 @@
+"""
+TBAC (Task-Based Access Control) Configuration
+"""
+import os
+import logging
+from dotenv import load_dotenv
+
+try:
+    from identityservice.sdk import IdentityServiceSdk
+    TBAC_AVAILABLE = True
+except ImportError:
+    TBAC_AVAILABLE = False
+    logging.warning("identityservice.sdk not available - TBAC disabled")
+
+
+logger = logging.getLogger(__name__)
+
+
+class TBAC:
+    """TBAC configuration and authorization handler"""
+
+    def __init__(self):
+        load_dotenv()
+
+        # TBAC credentials
+        self.client_api_key = os.getenv('CLIENT_AGENT_API_KEY')
+        self.client_id = os.getenv('CLIENT_AGENT_ID')
+        self.a2a_api_key = os.getenv('A2A_SERVICE_API_KEY')
+        self.a2a_id = os.getenv('A2A_SERVICE_ID')
+
+        self.client_sdk = None
+        self.a2a_sdk = None
+        self.client_authorized = False
+        self.a2a_authorized = False
+        self.client_token = None
+        self.a2a_token = None
+
+        if TBAC_AVAILABLE:
+            self._setup()
+
+    def _setup(self):
+        """Initialize TBAC SDKs"""
+        if not all([self.client_api_key, self.client_id, self.a2a_api_key, self.a2a_id]):
+            logger.warning("TBAC Disabled: Missing credentials")
+            return
+
+        try:
+            self.client_sdk = IdentityServiceSdk(api_key=self.client_api_key)
+            self.a2a_sdk = IdentityServiceSdk(api_key=self.a2a_api_key)
+            logger.info("TBAC SDKs initialized successfully")
+        except Exception as e:
+            logger.error(f"TBAC setup failed: {e}")
+
+    def authorize_client_to_a2a(self):
+        """Authorize client agent to communicate with A2A service"""
+        if not self.client_sdk or not self.a2a_sdk:
+            return True  # Skip if not configured
+
+        try:
+            # Get access token for client to access A2A service
+            self.client_token = self.client_sdk.access_token(self.a2a_id)
+
+            if not self.client_token:
+                logger.error("TBAC FAILED: Could not get client agent token")
+                return False
+
+            # Authorize the token with A2A service
+            self.client_authorized = self.a2a_sdk.authorize(self.client_token)
+
+            if self.client_authorized:
+                logger.info("TBAC: Client agent authorized to A2A service")
+                return True
+            else:
+                logger.error("TBAC FAILED: Client agent not authorized by A2A service")
+                return False
+
+        except Exception as e:
+            logger.error(f"TBAC client-to-a2a authorization failed: {e}")
+            return False
+
+    def authorize_a2a_to_client(self):
+        """Authorize A2A service to communicate with client agent"""
+        if not self.client_sdk or not self.a2a_sdk:
+            return True  # Skip if not configured
+
+        try:
+            # Get access token for A2A service to access client
+            self.a2a_token = self.a2a_sdk.access_token(self.client_id)
+
+            if not self.a2a_token:
+                logger.error("TBAC FAILED: Could not get A2A service token")
+                return False
+
+            # Authorize the token with client agent
+            self.a2a_authorized = self.client_sdk.authorize(self.a2a_token)
+
+            if self.a2a_authorized:
+                logger.info("TBAC: A2A service authorized to client agent")
+                return True
+            else:
+                logger.error("TBAC FAILED: A2A service not authorized by client agent")
+                return False
+
+        except Exception as e:
+            logger.error(f"TBAC A2A-to-client authorization failed: {e}")
+            return False
+
+    def authorize_bidirectional(self):
+        """Perform bidirectional authorization"""
+        client_to_a2a = self.authorize_client_to_a2a()
+        a2a_to_client = self.authorize_a2a_to_client()
+        return client_to_a2a and a2a_to_client
+
+    def is_client_authorized(self):
+        """Check if client agent is authorized to communicate with A2A service"""
+        return self.client_authorized or not all([self.client_api_key, self.a2a_api_key])
+
+    def is_a2a_authorized(self):
+        """Check if A2A service is authorized to communicate with client agent"""
+        return self.a2a_authorized or not all([self.client_api_key, self.a2a_api_key])
+
+    def is_fully_authorized(self):
+        """Check if both directions are authorized"""
+        return self.is_client_authorized() and self.is_a2a_authorized()
+
+    def is_voice_authorized(self):
+        """Check if voice agent is authorized to call A2A service"""
+        return self.is_client_authorized()
+
+    def check_voice_authorization(self):
+        """Block if voice agent not authorized - raises exception"""
+        if not self.is_voice_authorized():
+            raise PermissionError("TBAC: Voice agent not authorized to communicate with A2A service")
+
+    def check_a2a_authorization(self):
+        """Block if A2A service not authorized - raises exception"""
+        if not self.is_a2a_authorized():
+            raise PermissionError("TBAC: A2A service not authorized to handle messages")

agentic-healthcare-booking-app/requirements.txt

@@ -0,0 +1,30 @@
+# Core dependencies
+requests>=2.31.0
+python-dotenv>=1.0.0
+
+# Audio dependencies (optional but recommended)
+SpeechRecognition>=3.10.0
+PyAudio>=0.2.13
+pygame>=2.5.0
+gTTS>=2.4.0
+
+# Development dependencies
+pytest>=7.4.0
+pytest-asyncio>=0.21.0
+black>=23.0.0
+flake8>=6.0.0
+mypy>=1.5.0
+
+# a2a protocol
+a2a-sdk
+
+# agntcy
+ioa_observe_sdk
+agntcy-identity-service-sdk==0.0.7
+
+# a2a service
+Flask
+Flask-cors
+pydantic
+pytest-cov
+pytest-mock

agentic-healthcare-booking-app/setup.py

@@ -0,0 +1,11 @@
+from gettext import install
+from setuptools import setup, find_packages
+
+setup(
+    name="agentic-healthcare-booking-app",
+    packages=find_packages(),
+)
+
+
+# run "pip install -r requirements.txt" for installing dependency libraries
+# run "pip install -e ." for clean imports setup

agentic-healthcare-booking-app/triage_agent/infermedica/no_agntcy/client/triage_client.py

@@ -0,0 +1,167 @@
+"""
+External triage API client with timing
+"""
+import base64
+import time
+import logging
+import requests
+
+logger = logging.getLogger(__name__)
+
+
+class TriageAPIClient:
+    """Handles all external triage API communication"""
+
+    def __init__(self, config):
+        self.triage_app_id = config['triage_app_id']
+        self.triage_app_key = config['triage_app_key']
+        self.triage_instance_id = config['triage_instance_id']
+        self.triage_token_url = config['triage_token_url']
+        self.triage_base_url = config['triage_base_url']
+
+    def timed_external_request(self, method, url, description, **kwargs):
+        """Make a timed request to external API with detailed logging"""
+        start_time = time.time()
+        timestamp = time.strftime("%H:%M:%S", time.localtime(start_time))
+        logger.info(f"A2A-SERVICE: [{timestamp}] >>> {method} {description}")
+        logger.info(f"A2A-SERVICE: URL: {url}")
+
+        try:
+            if method == 'GET':
+                response = requests.get(url, **kwargs)
+            elif method == 'POST':
+                response = requests.post(url, **kwargs)
+            else:
+                raise ValueError(f"Unsupported method: {method}")
+
+            elapsed = time.time() - start_time
+            end_timestamp = time.strftime("%H:%M:%S", time.localtime())
+            elapsed_ms = elapsed * 1000
+
+            logger.info(f"A2A-SERVICE: [{end_timestamp}] <<< {response.status_code} | {elapsed:.3f}s ({elapsed_ms:.0f}ms)")
+
+            if response.status_code != 200:
+                logger.error(f"A2A-SERVICE: Error response: {response.text[:300]}")
+            else:
+                logger.info(f"A2A-SERVICE: Success - response length: {len(response.text)} chars")
+
+            return response, elapsed
+
+        except Exception as e:
+            elapsed = time.time() - start_time
+            end_timestamp = time.strftime("%H:%M:%S", time.localtime())
+            elapsed_ms = elapsed * 1000
+            logger.error(f"A2A-SERVICE: [{end_timestamp}] <<< ERROR: {e} | {elapsed:.3f}s ({elapsed_ms:.0f}ms)")
+            raise e
+
+    def get_triage_token(self):
+        """Get authentication token from external triage API with timing"""
+        logger.info("Requesting triage API authentication token")
+
+        creds = base64.b64encode(f"{self.triage_app_id}:{self.triage_app_key}".encode()).decode()
+        headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Basic {creds}",
+            "instance-id": self.triage_instance_id
+        }
+        payload = {"grant_type": "client_credentials"}
+
+        response, elapsed = self.timed_external_request(
+            'POST', self.triage_token_url, "Get OAuth Token",
+            headers=headers, json=payload, timeout=30
+        )
+
+        if response.status_code == 200:
+            token = response.json()['access_token']
+            logger.info(f"Successfully obtained triage API token")
+            return token
+
+        raise Exception(f"Failed to get token: {response.status_code} - {response.text}")
+
+    def create_triage_survey(self, token, age, sex):
+        """Create a new triage survey with timing"""
+        logger.info(f"Creating triage survey - age={age}, sex={sex}")
+
+        headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json"
+        }
+        payload = {
+            "sex": sex.lower(),
+            "age": {"value": age, "unit": "year"}
+        }
+
+        response, elapsed = self.timed_external_request(
+            'POST', f"{self.triage_base_url}/surveys", "Create Survey",
+            headers=headers, json=payload, timeout=30
+        )
+
+        if response.status_code == 200:
+            survey_id = response.json()['survey_id']
+            logger.info(f"Successfully created triage survey: {survey_id}")
+            return survey_id
+
+        raise Exception(f"Failed to create survey: {response.status_code} - {response.text}")
+
+    def send_triage_api_message(self, token, survey_id, message):
+        """Send message to external triage API with timing"""
+        logger.info(f"Sending message to triage API: '{message[:50]}...'")
+
+        headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json"
+        }
+        payload = {"user_message": message}
+
+        response, elapsed = self.timed_external_request(
+            'POST', f"{self.triage_base_url}/surveys/{survey_id}/messages",
+            "Send Message",
+            headers=headers, json=payload, timeout=30
+        )
+
+        if response.status_code == 200:
+            data = response.json()
+            external_state = data.get('survey_state', 'in_progress')
+            agent_response = data.get('assistant_message', '')
+
+            logger.info(f"Triage state: {external_state}")
+            logger.info(f"Triage response length: {len(agent_response)} chars")
+
+            return {
+                "success": True,
+                "response": agent_response,
+                "state": external_state
+            }
+        else:
+            logger.error(f"Triage API error: {response.status_code} - {response.text}")
+            return {
+                "success": False,
+                "response": "I'm having trouble with the medical assessment system."
+            }
+
+    def get_triage_summary(self, token, survey_id):
+        """Get triage summary from external API with timing"""
+        try:
+            headers = {"Authorization": f"Bearer {token}"}
+
+            response, elapsed = self.timed_external_request(
+                'GET', f"{self.triage_base_url}/surveys/{survey_id}/summary",
+                "Get Triage Summary",
+                headers=headers, timeout=30
+            )
+
+            if response.status_code == 200:
+                data = response.json()
+                logger.info(f"Triage summary retrieved successfully")
+                return {
+                    'success': True,
+                    'urgency_level': data.get('urgency', 'standard'),
+                    'doctor_type': data.get('doctor_type', 'general practitioner'),
+                    'notes': data.get('notes', 'Assessment completed')
+                }
+            else:
+                logger.warning(f"Failed to get triage summary: {response.status_code}")
+                return {'success': False}
+        except Exception as e:
+            logger.error(f"Error getting triage summary: {e}", exc_info=True)
+            return {'success': False}