Spaces:
Sleeping
Sleeping
File size: 4,693 Bytes
6ba374a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | # Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# add docstring to this module
import os
from datetime import datetime
from dotenv import load_dotenv
from google.adk.agents import Agent
from google.adk.tools import google_search
from google.adk.tools.agent_tool import AgentTool
from google.adk.tools.langchain_tool import LangchainTool
try:
from google.adk.tools.mcp_tool import MCPToolset, StreamableHTTPConnectionParams
except ImportError:
try:
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
from google.adk.tools.mcp_tool.mcp_toolset import StreamableHTTPConnectionParams
except ImportError:
MCPToolset = None
StreamableHTTPConnectionParams = None
from langchain_community.tools import StackExchangeTool
from langchain_community.utilities import StackExchangeAPIWrapper
# Load environment variables
load_dotenv()
# ----- Example of a Function tool -----
def get_current_date() -> dict:
"""
Get the current date in the format YYYY-MM-DD
"""
return {"current_date": datetime.now().strftime("%Y-%m-%d")}
# ----- Example of a Built-in Tool -----
search_agent = Agent(
model="gemini-2.5-flash",
name="search_agent",
instruction="""
You're a specialist in Google Search.
""",
tools=[google_search],
)
search_tool = AgentTool(search_agent)
# ----- Example of a Third Party Tool (LangChainTool) -----
# Initialize StackExchange tool with error handling for connection timeouts
langchain_tool = None
try:
# Try to initialize StackExchange tool
# Note: StackExchangeAPIWrapper may attempt to connect during initialization
# If connection times out, we'll skip it and continue without it
stack_exchange_wrapper = StackExchangeAPIWrapper()
stack_exchange_tool = StackExchangeTool(api_wrapper=stack_exchange_wrapper)
langchain_tool = LangchainTool(stack_exchange_tool)
except (ConnectionError, TimeoutError, Exception) as e:
# StackExchange API not available (timeout, network issue, etc.)
# Continue without it - it's not critical for core functionality
# Only print warning if it's not a timeout (to avoid spam)
error_str = str(e).lower()
if "timeout" not in error_str and "connect" not in error_str:
print(f"Warning: StackExchange tool initialization failed: {e}")
# Silently continue without StackExchange tool
langchain_tool = None
# Use local functions instead of the standalone MCP server to avoid Docker dependency issues
from .local_tools import (
get_tickets_by_status,
get_tickets_by_priority,
get_ticket_by_id,
get_tickets_by_assignee,
search_tickets,
create_ticket,
update_ticket_status,
update_ticket_priority,
get_tickets_by_date_range
)
toolbox_tools = [
get_tickets_by_status,
get_tickets_by_priority,
get_ticket_by_id,
get_tickets_by_assignee,
search_tickets,
create_ticket,
update_ticket_status,
update_ticket_priority,
get_tickets_by_date_range
]
# ----- Example of an MCP Tool (streamable-http) -----
# DISABLED: MCP tools cause runtime errors when server is unreachable
# The GitHub MCP server requires a valid GITHUB_PERSONAL_ACCESS_TOKEN
# and causes the entire ADK agent to fail if connection times out.
# Using local tools for GitHub search instead (in local_tools.py)
mcp_tools = None
# Uncomment below if you have a valid GitHub token and want MCP tools:
# github_token = os.getenv("GITHUB_PERSONAL_ACCESS_TOKEN")
# if github_token and len(github_token) > 10:
# try:
# mcp_tools = MCPToolset(
# connection_params=StreamableHTTPConnectionParams(
# url="https://api.githubcopilot.com/mcp/",
# headers={"Authorization": f"Bearer {github_token}"},
# ),
# tool_filter=[
# "search_repositories",
# "search_issues",
# "list_issues",
# "get_issue",
# "list_pull_requests",
# "get_pull_request",
# ],
# )
# except Exception as e:
# print(f"GitHub MCP tools not available: {e}")
# mcp_tools = None
|