Create app.py

app.py

@@ -0,0 +1,355 @@
+import gradio as gr
+import openai
+import requests
+import json
+import os
+from typing import Dict, List, Any
+from datetime import datetime
+from pydantic import BaseModel
+
+# Set up OpenAI client
+openai.api_key = os.getenv("OPENAI_API_KEY")
+client = openai.OpenAI()
+
+# SAP API Configuration
+SAP_API_KEY = os.getenv("SAP_API_KEY")
+BASE_URL = "https://sandbox.api.sap.com/s4hanacloud/sap/opu/odata/sap/API_BUSINESS_PARTNER"
+
+class SAPBusinessPartnerAgent:
+    def __init__(self):
+        self.conversation_history = []
+    
+    def parse_search_query(self, query: str) -> Dict[str, Any]:
+        """Parse natural language query into SAP API parameters using OpenAI"""
+        system_prompt = """
+        Parse this business partner search query into SAP OData parameters.
+        
+        Available fields for A_BusinessPartner:
+        - BusinessPartner (ID)
+        - BusinessPartnerFullName (Name)
+        - BusinessPartnerCategory (1=Person, 2=Organization, 3=Group)
+        - BusinessPartnerGrouping
+        - CreationDate
+        - IsMarkedForArchiving
+        - SearchTerm1, SearchTerm2
+        
+        Return JSON with OData query parameters.
+        
+        Examples:
+        - "Find customer ABC Corp" → {"$filter": "contains(BusinessPartnerFullName,'ABC Corp')", "$top": "10"}
+        - "Show me organizations" → {"$filter": "BusinessPartnerCategory eq '2'", "$top": "10"}
+        - "Active partners only" → {"$filter": "IsMarkedForArchiving eq false", "$top": "10"}
+        
+        Always include $top with a reasonable limit (max 20).
+        """
+        
+        try:
+            response = client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": system_prompt},
+                    {"role": "user", "content": f"Query: {query}"}
+                ],
+                response_format={"type": "json_object"}
+            )
+            
+            result = json.loads(response.choices[0].message.content)
+            # Ensure we have basic parameters
+            if "$top" not in result:
+                result["$top"] = "10"
+            return result
+            
+        except Exception as e:
+            # Fallback to basic search
+            return {
+                "$filter": f"contains(BusinessPartnerFullName,'{query}')",
+                "$top": "10"
+            }
+    
+    def call_sap_api(self, entity_set: str, params: Dict[str, Any]) -> Dict[str, Any]:
+        """Execute SAP API call"""
+        url = f"{BASE_URL}/{entity_set}"
+        
+        # Build query parameters
+        query_params = {
+            "$format": "json",
+            "$inlinecount": "allpages"
+        }
+        
+        # Add user parameters
+        for key, value in params.items():
+            query_params[key] = value
+        
+        headers = {
+            "Accept": "application/json",
+            "APIKey": SAP_API_KEY or "demo_key"
+        }
+        
+        try:
+            response = requests.get(url, params=query_params, headers=headers, timeout=30)
+            
+            if response.status_code == 200:
+                return response.json()
+            else:
+                # Return mock data for demo if API fails
+                return self.get_mock_data(entity_set)
+                
+        except Exception as e:
+            # Return mock data for demo
+            return self.get_mock_data(entity_set)
+    
+    def get_mock_data(self, entity_set: str) -> Dict[str, Any]:
+        """Return mock SAP data for demo purposes"""
+        mock_data = {
+            "d": {
+                "results": [
+                    {
+                        "BusinessPartner": "1000000001",
+                        "BusinessPartnerFullName": "Demo Customer Corp",
+                        "BusinessPartnerCategory": "2",
+                        "CreationDate": "/Date(1640995200000)/",
+                        "IsMarkedForArchiving": False,
+                        "SearchTerm1": "DEMO"
+                    },
+                    {
+                        "BusinessPartner": "1000000002", 
+                        "BusinessPartnerFullName": "Sample Supplier Ltd",
+                        "BusinessPartnerCategory": "2",
+                        "CreationDate": "/Date(1641081600000)/",
+                        "IsMarkedForArchiving": False,
+                        "SearchTerm1": "SAMPLE"
+                    },
+                    {
+                        "BusinessPartner": "1000000003",
+                        "BusinessPartnerFullName": "Test Industries Inc",
+                        "BusinessPartnerCategory": "2", 
+                        "CreationDate": "/Date(1641168000000)/",
+                        "IsMarkedForArchiving": False,
+                        "SearchTerm1": "TEST"
+                    }
+                ],
+                "__count": "3"
+            }
+        }
+        return mock_data
+    
+    def format_business_partner_response(self, response: Dict, original_query: str) -> str:
+        """Format business partner API response into readable text"""
+        try:
+            if 'd' in response and 'results' in response['d']:
+                results = response['d']['results']
+                total_count = response['d'].get('__count', len(results))
+                
+                if not results:
+                    return f"❌ No business partners found for: '{original_query}'"
+                
+                # Format the response
+                formatted_response = f"## 📊 Business Partner Search Results\n\n"
+                formatted_response += f"**Query:** {original_query}\n"
+                formatted_response += f"**Found:** {total_count} business partner(s)\n\n"
+                
+                for i, partner in enumerate(results, 1):
+                    bp_id = partner.get('BusinessPartner', 'N/A')
+                    bp_name = partner.get('BusinessPartnerFullName', 'N/A')
+                    bp_category = partner.get('BusinessPartnerCategory', 'N/A')
+                    is_archived = partner.get('IsMarkedForArchiving', False)
+                    
+                    category_text = {
+                        '1': '👤 Person',
+                        '2': '🏢 Organization',
+                        '3': '🏭 Group'
+                    }.get(bp_category, '❓ Unknown')
+                    
+                    status = '🟢 Active' if not is_archived else '🔴 Archived'
+                    
+                    formatted_response += f"### {i}. {bp_name}\n"
+                    formatted_response += f"- **ID:** {bp_id}\n"
+                    formatted_response += f"- **Type:** {category_text}\n"
+                    formatted_response += f"- **Status:** {status}\n\n"
+                
+                # Add helpful suggestions
+                formatted_response += "---\n"
+                formatted_response += "💡 **What you can do next:**\n"
+                formatted_response += "- Ask for specific partner details\n"
+                formatted_response += "- Search for customers or suppliers\n"
+                formatted_response += "- Find partners by location or industry\n"
+                
+                return formatted_response
+            else:
+                return f"⚠️ Received unexpected response format for: '{original_query}'"
+                
+        except Exception as e:
+            return f"❌ Error formatting response: {str(e)}"
+    
+    def search_business_partners(self, query: str) -> str:
+        """Main search function for business partners"""
+        try:
+            # Parse the query
+            params = self.parse_search_query(query)
+            
+            # Call SAP API
+            response = self.call_sap_api("A_BusinessPartner", params)
+            
+            # Format response
+            return self.format_business_partner_response(response, query)
+            
+        except Exception as e:
+            return f"❌ Error searching business partners: {str(e)}"
+    
+    def process_user_query(self, user_query: str, history: List) -> tuple:
+        """Process user query and return response with updated history"""
+        
+        # Add user message to history
+        history.append([user_query, None])
+        
+        try:
+            # Determine intent and process accordingly
+            if any(keyword in user_query.lower() for keyword in ['search', 'find', 'show', 'list', 'get']):
+                response = self.search_business_partners(user_query)
+            else:
+                # Use OpenAI to understand intent and provide guidance
+                system_prompt = """
+                You are a SAP Business Partner Assistant. The user is asking about business partner management.
+                
+                If they're asking for:
+                - Search/find operations: Guide them to use search terms
+                - Help/guidance: Explain what you can do
+                - Specific features: Explain SAP Business Partner capabilities
+                
+                Be helpful and guide them to ask specific questions about business partners.
+                """
+                
+                ai_response = client.chat.completions.create(
+                    model="gpt-4o-mini",
+                    messages=[
+                        {"role": "system", "content": system_prompt},
+                        {"role": "user", "content": user_query}
+                    ]
+                )
+                
+                response = ai_response.choices[0].message.content
+            
+            # Update history with response
+            history[-1][1] = response
+            
+        except Exception as e:
+            error_response = f"❌ Sorry, I encountered an error: {str(e)}\n\nTry asking: 'Find business partners with name Demo'"
+            history[-1][1] = error_response
+        
+        return "", history
+
+# Initialize the agent
+sap_agent = SAPBusinessPartnerAgent()
+
+# Create Gradio interface
+def create_interface():
+    with gr.Blocks(
+        title="SAP Business Partner Agent",
+        theme=gr.themes.Soft(),
+        css="""
+        .gradio-container {
+            max-width: 1200px !important;
+        }
+        .agent-header {
+            text-align: center;
+            background: linear-gradient(90deg, #0070f3, #00d4ff);
+            color: white;
+            padding: 20px;
+            border-radius: 10px;
+            margin-bottom: 20px;
+        }
+        """
+    ) as demo:
+        
+        # Header
+        gr.HTML("""
+        <div class="agent-header">
+            <h1>🤖 SAP Business Partner Agent</h1>
+            <p>Intelligent assistant for SAP Business Partner master data management</p>
+        </div>
+        """)
+        
+        # Main chat interface
+        with gr.Row():
+            with gr.Column(scale=4):
+                chatbot = gr.Chatbot(
+                    height=500,
+                    label="Chat with SAP Business Partner Agent",
+                    placeholder="Start by asking about business partners..."
+                )
+                
+                with gr.Row():
+                    msg = gr.Textbox(
+                        placeholder="Ask me about business partners (e.g., 'Find partners with name Demo')",
+                        label="Your Message",
+                        scale=4
+                    )
+                    submit_btn = gr.Button("Send", variant="primary", scale=1)
+                
+                # Example buttons
+                gr.Markdown("### 💡 Quick Examples:")
+                with gr.Row():
+                    example1 = gr.Button("Find Demo partners", size="sm")
+                    example2 = gr.Button("Show organizations", size="sm") 
+                    example3 = gr.Button("List active partners", size="sm")
+                    clear_btn = gr.Button("Clear Chat", size="sm", variant="secondary")
+            
+            # Sidebar with information
+            with gr.Column(scale=1):
+                gr.Markdown("""
+                ### 🔧 Agent Capabilities
+                
+                **Search & Retrieve:**
+                - Find business partners by name
+                - Filter by partner type
+                - Search by various criteria
+                
+                **Partner Types:**
+                - 👤 Persons
+                - 🏢 Organizations 
+                - 🏭 Groups
+                
+                **Data Sources:**
+                - SAP Business Partner API
+                - Real-time SAP data
+                - Customer information
+                - Supplier details
+                
+                ### 📝 Sample Queries
+                - "Find partners named ABC"
+                - "Show me all organizations"
+                - "List active business partners"
+                - "Search for suppliers"
+                """)
+        
+        # Event handlers
+        msg.submit(
+            sap_agent.process_user_query,
+            inputs=[msg, chatbot],
+            outputs=[msg, chatbot]
+        )
+        
+        submit_btn.click(
+            sap_agent.process_user_query,
+            inputs=[msg, chatbot], 
+            outputs=[msg, chatbot]
+        )
+        
+        # Example button events
+        example1.click(lambda: "Find partners with name Demo", outputs=msg)
+        example2.click(lambda: "Show me organizations", outputs=msg)
+        example3.click(lambda: "List active partners only", outputs=msg)
+        clear_btn.click(lambda: [], outputs=chatbot)
+        
+        # Footer
+        gr.Markdown("""
+        ---
+        **SAP Business Partner Agent** | Powered by OpenAI & SAP APIs | Built for Hugging Face Spaces
+        """)
+    
+    return demo
+
+# Launch the app
+if __name__ == "__main__":
+    demo = create_interface()
+    demo.launch()