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)