Create mcp.txt

mcp.txt

@@ -0,0 +1,232 @@
+import gradio as gr
+import os
+import asyncio
+from typing import List
+from google import genai
+from google.genai import types
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+# MCP-Server Konfiguration
+server_params = StdioServerParameters(
+    command="npx",
+    args=[
+        "mcp-remote",
+        "https://mgokg-db-timetable-api.hf.space/gradio_api/mcp/"
+    ]
+)
+
+async def generate1(input_text):
+    try:
+        client = genai.Client(api_key=os.environ.get("GEMINI_API_KEY"))
+    except Exception as e:
+        return f"Fehler bei der Initialisierung: {e}", ""
+    
+    model = "gemini-2.5-flash"
+    
+    try:
+        # MCP-Session aufbauen
+        async with stdio_client(server_params) as (read, write):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+                
+                # MCP-Tools abrufen
+                mcp_tools_data = await session.list_tools()
+                
+                # Tools in Gemini-kompatibles Format konvertieren
+                function_declarations = []
+                
+                # 1. MCP Tools hinzufügen
+                for tool in mcp_tools_data.tools:
+                    schema = tool.inputSchema if tool.inputSchema else {
+                        "type": "object",
+                        "properties": {},
+                        "required": []
+                    }
+                    
+                    function_declarations.append(
+                        types.FunctionDeclaration(
+                            name=tool.name,
+                            description=tool.description or "MCP Tool",
+                            parameters=schema
+                        )
+                    )
+                
+                # 2. Google Search Tool hinzufügen (als Function Declaration)
+                # Die Gemini API erwartet Google Search als spezielles Tool
+                google_search_declaration = types.FunctionDeclaration(
+                    name="google_search",
+                    description="Search the internet for current information, news, facts, and general knowledge. Use this for questions about current events, weather, general knowledge, or when you need up-to-date information from the web.",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "query": {
+                                "type": "string",
+                                "description": "The search query to look up on Google"
+                            }
+                        },
+                        "required": ["query"]
+                    }
+                )
+                function_declarations.append(google_search_declaration)
+                
+                # Alle Tools kombinieren
+                tools = [types.Tool(function_declarations=function_declarations)]
+                
+                # System Prompt zur Tool-Auswahl
+                system_instruction = """You are a helpful assistant with access to two types of tools:
+
+1. TRAIN CONNECTION TOOLS (MCP): Use these ONLY when the user asks about train connections, schedules, routes, or German railway (DB) information. Available tools: """ + ", ".join([t.name for t in mcp_tools_data.tools]) + """
+
+2. GOOGLE SEARCH: Use this for ALL other queries - general knowledge, current events, weather, news, calculations, or any topic not related to train connections.
+
+Choose the appropriate tool based on the user's intent. For train connections, use the MCP tools. For everything else, use Google Search."""
+
+                contents = [types.Content(
+                    role="user", 
+                    parts=[types.Part.from_text(text=input_text)]
+                )]
+                
+                # Erster API-Aufruf
+                response = await client.aio.models.generate_content(
+                    model=model,
+                    contents=contents,
+                    config=types.GenerateContentConfig(
+                        tools=tools,
+                        temperature=0.4,
+                        system_instruction=system_instruction
+                    )
+                )
+                
+                # Agentic Loop für Tool-Calls
+                turn_count = 0
+                max_turns = 5
+                
+                while response.candidates and turn_count < max_turns:
+                    candidate = response.candidates[0]
+                    
+                    if not hasattr(candidate, 'content') or not candidate.content:
+                        break
+                        
+                    function_calls = [
+                        part for part in candidate.content.parts 
+                        if hasattr(part, 'function_call') and part.function_call
+                    ]
+                    
+                    if not function_calls:
+                        break
+                    
+                    turn_count += 1
+                    contents.append(candidate.content)
+                    tool_responses = []
+                    
+                    for part in function_calls:
+                        fc = part.function_call
+                        tool_name = fc.name
+                        tool_args = dict(fc.args) if fc.args else {}
+                        
+                        try:
+                            # Unterscheide zwischen MCP-Tools und Google Search
+                            if tool_name == "google_search":
+                                # Google Search über Gemini's eingebautes Tool
+                                # Wir simulieren hier den Aufruf durch einen neuen Request mit google_search Tool
+                                search_response = await client.aio.models.generate_content(
+                                    model=model,
+                                    contents=[types.Content(
+                                        role="user",
+                                        parts=[types.Part.from_text(text=f"Search Google for: {tool_args.get('query', '')}")]
+                                    )],
+                                    config=types.GenerateContentConfig(
+                                        tools=[types.Tool(google_search=types.GoogleSearch())],
+                                        temperature=0.4
+                                    )
+                                )
+                                result_text = search_response.text
+                                
+                            else:
+                                # MCP-Tool ausführen
+                                tool_result = await session.call_tool(tool_name, tool_args)
+                                
+                                if tool_result.isError:
+                                    result_text = f"Error: {tool_result.content[0].text if tool_result.content else 'Unknown error'}"
+                                else:
+                                    result_text = tool_result.content[0].text if tool_result.content else "No result"
+                            
+                            tool_responses.append(
+                                types.FunctionResponse(
+                                    name=tool_name,
+                                    response={"result": result_text}
+                                )
+                            )
+                            
+                        except Exception as e:
+                            tool_responses.append(
+                                types.FunctionResponse(
+                                    name=tool_name,
+                                    response={"error": str(e)}
+                                )
+                            )
+                    
+                    contents.append(types.Content(role="user", parts=[
+                        types.Part.from_function_response(
+                            name=tr.name,
+                            response=tr.response
+                        ) for tr in tool_responses
+                    ]))
+                    
+                    # Nächster API-Aufruf
+                    response = await client.aio.models.generate_content(
+                        model=model,
+                        contents=contents,
+                        config=types.GenerateContentConfig(
+                            tools=tools, 
+                            temperature=0.4,
+                            system_instruction=system_instruction
+                        )
+                    )
+                
+                return response.text, ""
+    
+    except Exception as e:
+        import traceback
+        return f"Fehler während der Verarbeitung: {str(e)}\n\n{traceback.format_exc()}", ""
+
+# Gradio UI Wrapper
+def generate(input_text):
+    try:
+        return asyncio.run(generate1(input_text))
+    except Exception as e:
+        import traceback
+        return f"UI Fehler: {str(e)}\n\n{traceback.format_exc()}", ""
+
+if __name__ == '__main__':
+    with gr.Blocks() as demo:
+        gr.Markdown("# Gemini 2.5 Flash + Google Search + DB Timetable (MCP)")
+        
+        with gr.Row():
+            with gr.Column():
+                input_textbox = gr.Textbox(
+                    lines=3, 
+                    label="Anfrage", 
+                    placeholder="z.B. Wie komme ich von Berlin nach Hamburg? oder Was ist die Hauptstadt von Frankreich?"
+                )
+                submit_button = gr.Button("Senden", variant="primary")
+            
+            with gr.Column():
+                output_textbox = gr.Markdown(label="Antwort")
+                status_textbox = gr.Textbox(label="Status", interactive=False)
+        
+        submit_button.click(
+            fn=generate,
+            inputs=input_textbox,
+            outputs=[output_textbox, status_textbox]
+        )
+        
+        # Enter-Taste unterstützung
+        input_textbox.submit(
+            fn=generate,
+            inputs=input_textbox,
+            outputs=[output_textbox, status_textbox]
+        )
+    
+    demo.launch(show_error=True, share=False)