import gradio as gr
from huggingface_hub import InferenceClient
import json
import os
import sys
import subprocess
import time
from pathlib import Path
import re
import urllib.request
import urllib.error
def start_mcp_service(hf_token=None):
"""Start the MCP service with stdio transport"""
try:
# Change to the MCP_Financial_Report directory
# Try HF path first, then local path
mcp_dirs = [
"/app/MCP_Financial_Report", # HF environment path
os.path.join(os.path.dirname(__file__), "MCP_Financial_Report") # Local path
]
mcp_dir = None
for dir_path in mcp_dirs:
if os.path.exists(dir_path):
mcp_dir = dir_path
break
if not mcp_dir:
print(f"MCP directory not found in any of: {mcp_dirs}")
return False, None
# Use Python 3.10 explicitly to ensure MCP module is available
python_executable = "/usr/local/bin/python3.10"
if not os.path.exists(python_executable):
python_executable = sys.executable
print(f"Starting MCP service with Python: {python_executable}") # Debug information
print(f"MCP directory: {mcp_dir}") # Debug information
# Prepare environment variables for the subprocess
env = os.environ.copy()
# Pass HF token to subprocess if available
if hf_token and isinstance(hf_token, str):
env['HUGGING_FACE_HUB_TOKEN'] = hf_token
print(f"Passing HF token to MCP subprocess (length: {len(hf_token)})")
else:
print("WARNING: No HF token available for MCP subprocess")
# Start the MCP server as a subprocess with stdio
mcp_process = subprocess.Popen([
python_executable, "financial_mcp_server.py"
],
cwd=mcp_dir,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
stdin=subprocess.PIPE,
bufsize=0, # Unbuffered
env=env # Pass environment variables including HF token
)
# Wait for the server to start
time.sleep(2)
# Check if the process is still running
if mcp_process.poll() is not None:
stdout, stderr = mcp_process.communicate()
print(f"MCP service failed to start:")
print(f"STDOUT: {stdout.decode()}")
print(f"STDERR: {stderr.decode()}")
return False, None
print("MCP service started successfully")
# Reset the global initialization flag for the new process
global MCP_INITIALIZED
MCP_INITIALIZED = False
return True, mcp_process
except Exception as e:
print(f"Error starting MCP service: {e}")
return False, None
def initialize_mcp_session_stdio(mcp_process):
"""Initialize MCP session via stdio"""
try:
# Create the initialize request
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2025-06-18",
"capabilities": {
"experimental": {},
"sampling": {},
"elicitations": {},
"roots": {}
},
"clientInfo": {
"name": "gradio-client",
"version": "1.0.0"
}
}
}
print(f"Sending MCP initialize request: {request}") # Debug information
# Send request to MCP process via stdin
request_str = json.dumps(request) + "\n"
mcp_process.stdin.write(request_str.encode('utf-8'))
mcp_process.stdin.flush()
# Read response from MCP process via stdout
response_data = b""
timeout = time.time() + 10 # 10 second timeout
while time.time() < timeout:
# Read a byte at a time until we get a newline
byte = mcp_process.stdout.read(1)
if not byte:
# Check if process is still alive
if mcp_process.poll() is not None:
raise Exception("MCP process terminated unexpectedly during initialization")
time.sleep(0.01) # Small delay to prevent busy waiting
continue
response_data += byte
if byte == b'\n':
break
else:
raise Exception("Timeout waiting for MCP initialize response")
# Parse the response
response_str = response_data.decode('utf-8').strip()
if not response_str:
raise Exception("Empty response from MCP initialize")
print(f"MCP initialize response: {response_str}") # Debug information
response = json.loads(response_str)
# Check for error response
if "error" in response:
error_msg = f"MCP Error {response['error'].get('code', 'unknown')}: {response['error'].get('message', 'Unknown error')}"
raise Exception(error_msg)
# Check if initialization was successful
if "result" in response:
# Send initialized notification as required by MCP spec
initialized_notification = {
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
print(f"Sending MCP initialized notification: {initialized_notification}") # Debug information
# Send notification to MCP process via stdin
notification_str = json.dumps(initialized_notification) + "\n"
mcp_process.stdin.write(notification_str.encode('utf-8'))
mcp_process.stdin.flush()
return True
else:
raise Exception("MCP initialization failed: no result in response")
except json.JSONDecodeError as e:
raise Exception(f"Failed to parse MCP initialize response as JSON: {str(e)}")
except Exception as e:
print(f"Error initializing MCP session via stdio: {str(e)}")
raise e
# Configuration for EasyReportDataMCP service (本地服务)
import os
THIRD_PARTY_MCP_URL = "http://localhost:7861/messages" # 本地EasyReportDataMCP服务 (SSE transport /messages endpoint)
THIRD_PARTY_MCP_TOOLS = [
"search_company",
"get_company_info",
"get_company_filings",
"get_financial_data",
"extract_financial_metrics",
"get_latest_financial_data",
"advanced_search_company"
]
# ✅ Configuration for MarketandStockMCP service
MARKET_STOCK_MCP_URL = "http://localhost:7870/messages" # SSE transport /messages endpoint
MARKET_STOCK_MCP_TOOLS = [
"get_quote",
"get_market_news",
"get_company_news"
]
# Global variable to track MCP session initialization
MCP_INITIALIZED = False
THIRD_PARTY_MCP_INITIALIZED = False
MARKET_STOCK_MCP_INITIALIZED = False
def call_mcp_tool_stdio(mcp_process, tool_name, arguments):
"""
Call an MCP tool via stdio with proper error handling and validation
"""
global MCP_INITIALIZED
output_messages = []
try:
# Initialize if needed
if not MCP_INITIALIZED:
output_messages.append(f"Initializing MCP session for tool: {tool_name}")
success = initialize_mcp_session_stdio(mcp_process)
if not success:
raise Exception("Failed to initialize MCP session")
MCP_INITIALIZED = True
output_messages.append("MCP session initialized successfully")
else:
output_messages.append(f"Using existing MCP session for tool: {tool_name}")
# Create the request according to MCP specification
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments
}
}
output_messages.append(f"Sending MCP request: {request}")
print(f"[DEBUG] Sending MCP request: {request}") # Add debug print
# Send request to MCP process via stdin
request_str = json.dumps(request) + "\n"
mcp_process.stdin.write(request_str.encode('utf-8'))
mcp_process.stdin.flush()
# Read response from MCP process via stdout with timeout
response_data = b""
timeout = time.time() + 30 # 30 second timeout
while time.time() < timeout:
# Read a byte at a time until we get a newline
byte = mcp_process.stdout.read(1)
if not byte:
# Check if process is still alive
if mcp_process.poll() is not None:
error_msg = "MCP process terminated unexpectedly"
print(f"[DEBUG] {error_msg}") # Add debug print
raise Exception(error_msg)
time.sleep(0.01) # Small delay to prevent busy waiting
continue
response_data += byte
if byte == b'\n':
break
else:
# If we got here, we timed out
error_msg = f"Timeout waiting for MCP response for tool: {tool_name}"
print(f"[DEBUG] {error_msg}") # Add debug print
raise TimeoutError(error_msg)
# Parse the response
response_str = response_data.decode('utf-8').strip()
print(f"[DEBUG] Raw MCP response: {response_str}") # Add debug print
if not response_str:
error_msg = "Empty response from MCP tool"
print(f"[DEBUG] {error_msg}") # Add debug print
raise Exception(error_msg)
response = json.loads(response_str)
# Check for error response
if "error" in response:
error_msg = f"MCP Error {response['error'].get('code', 'unknown')}: {response['error'].get('message', 'Unknown error')}"
# Add additional context for common errors
if response['error'].get('code') == -32602:
error_msg += f". This typically means the arguments provided do not match the tool's expected input schema. Provided arguments: {json.dumps(arguments)}"
print(f"[DEBUG] {error_msg}") # Add debug print
raise Exception(error_msg)
# Return the result
if "result" in response:
result = response["result"]
print(f"[DEBUG] MCP tool result: {result}") # Add debug print
return result
else:
error_msg = "MCP tool call failed: no result in response"
print(f"[DEBUG] {error_msg}") # Add debug print
raise Exception(error_msg)
except json.JSONDecodeError as e:
error_msg = f"Failed to parse MCP response as JSON: {str(e)}"
print(f"[DEBUG] JSON decode error: {error_msg}") # Add debug print
raise Exception(error_msg)
except Exception as e:
output_messages.append(f"Error calling MCP tool {tool_name}: {str(e)}")
print(f"[DEBUG] Exception in call_mcp_tool_stdio: {str(e)}") # Add debug print
raise e
def call_third_party_mcp_tool(tool_name, arguments):
"""
Call a third-party MCP tool via HTTP with proper error handling
Note: Third-party MCP service doesn't require authentication
"""
import httpx
import asyncio
import time # Add timing
global THIRD_PARTY_MCP_INITIALIZED
output_messages = []
start_time = time.time() # Start timing
print(f"[TIMING] Starting third-party MCP call for {tool_name}")
try:
# Create the request according to MCP specification
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments
}
}
output_messages.append(f"Sending third-party MCP request: {request}")
print(f"[DEBUG] Sending third-party MCP request: {request}")
request_prep_time = time.time()
print(f"[TIMING] Request preparation took: {request_prep_time - start_time:.3f}s")
# Use httpx to call the third-party MCP service
async def make_request():
http_start = time.time()
async with httpx.AsyncClient(timeout=60.0) as client:
# ✅ 直接使用全局变量,不需要导入
# THIRD_PARTY_MCP_URL 已在文件顶部定义(第178行)
print(f"[TIMING] Target URL: {THIRD_PARTY_MCP_URL}")
print(f"[TIMING] Request payload: {request}")
print(f"[TIMING] HTTP client created, making POST request...")
post_start = time.time()
# Third-party MCP service doesn't require authentication
response = await client.post(
THIRD_PARTY_MCP_URL,
json=request,
headers={
"Content-Type": "application/json"
}
)
post_end = time.time()
print(f"[TIMING] HTTP POST took: {post_end - post_start:.3f}s")
print(f"[TIMING] Response status: {response.status_code}")
print(f"[TIMING] Response size: {len(response.content)} bytes")
response.raise_for_status()
result = response.json()
http_end = time.time()
print(f"[TIMING] Total HTTP operation took: {http_end - http_start:.3f}s")
return result
# Run async function
async_start = time.time()
print(f"[TIMING] Starting async execution...")
try:
# Try to get the current event loop
try:
loop = asyncio.get_running_loop()
print(f"[TIMING] Detected running event loop, applying nest_asyncio...")
nest_start = time.time()
# If we're already in an async context, we need to use nest_asyncio or run in thread
import nest_asyncio
nest_asyncio.apply()
nest_end = time.time()
print(f"[TIMING] nest_asyncio.apply() took: {nest_end - nest_start:.3f}s")
exec_start = time.time()
response = loop.run_until_complete(make_request())
exec_end = time.time()
print(f"[TIMING] run_until_complete() took: {exec_end - exec_start:.3f}s")
except RuntimeError:
print(f"[TIMING] No running loop, using asyncio.run()...")
# No running loop, create a new one
response = asyncio.run(make_request())
except ImportError:
print(f"[TIMING] nest_asyncio not available, creating new loop...")
# nest_asyncio not available, fall back to creating new loop
try:
loop = asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
response = loop.run_until_complete(make_request())
async_end = time.time()
print(f"[TIMING] Async execution completed in: {async_end - async_start:.3f}s")
print(f"[DEBUG] Third-party MCP response: {response}")
total_time = time.time() - start_time
print(f"[TIMING] ⏱️ TOTAL third-party MCP call for '{tool_name}' took: {total_time:.3f}s")
# Check if response is None
if response is None:
error_msg = "Third-party MCP tool call failed: received None response"
print(f"[DEBUG] {error_msg}")
raise Exception(error_msg)
# Check for error response (only if error field exists and is not None)
if isinstance(response, dict) and "error" in response and response["error"] is not None:
error_content = response["error"]
error_code = error_content.get('code', 'unknown') if isinstance(error_content, dict) else 'unknown'
error_message = error_content.get('message', 'Unknown error') if isinstance(error_content, dict) else str(error_content)
error_msg = f"Third-party MCP Error {error_code}: {error_message}"
print(f"[DEBUG] {error_msg}")
raise Exception(error_msg)
# Return the result
if isinstance(response, dict) and "result" in response:
result = response["result"]
print(f"[DEBUG] Third-party MCP tool result: {result}")
return result
else:
error_msg = "Third-party MCP tool call failed: no result in response"
print(f"[DEBUG] {error_msg}")
raise Exception(error_msg)
except json.JSONDecodeError as e:
error_msg = f"Failed to parse third-party MCP response as JSON: {str(e)}"
print(f"[DEBUG] JSON decode error: {error_msg}")
raise Exception(error_msg)
except Exception as e:
# Ensure we have a meaningful error message
error_str = str(e) if str(e) else "Unknown error occurred"
output_messages.append(f"Error calling third-party MCP tool {tool_name}: {error_str}")
print(f"[DEBUG] Exception in call_third_party_mcp_tool: {error_str}")
print(f"[DEBUG] Exception type: {type(e)}")
# Print full traceback for debugging
import traceback
print(f"[DEBUG] Full traceback: {traceback.format_exc()}")
raise Exception(error_str)
def call_market_stock_mcp_tool(tool_name, arguments):
"""
Call MarketandStockMCP tool via HTTP (SSE transport)
"""
import httpx
import asyncio
import time
global MARKET_STOCK_MCP_INITIALIZED
output_messages = []
start_time = time.time()
print(f"[TIMING] Starting MarketandStockMCP call for {tool_name}")
try:
# Create the request according to MCP specification
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments
}
}
output_messages.append(f"Sending MarketandStockMCP request: {request}")
print(f"[DEBUG] Sending MarketandStockMCP request: {request}")
request_prep_time = time.time()
print(f"[TIMING] Request preparation took: {request_prep_time - start_time:.3f}s")
# Use httpx to call the MarketandStockMCP service
async def make_request():
http_start = time.time()
async with httpx.AsyncClient(timeout=60.0) as client:
# ✅ 直接使用全局变量
# MARKET_STOCK_MCP_URL 已在文件顶部定义
print(f"[TIMING] Target URL: {MARKET_STOCK_MCP_URL}")
print(f"[TIMING] Request payload: {request}")
print(f"[TIMING] HTTP client created, making POST request...")
post_start = time.time()
# MarketandStockMCP uses SSE transport
response = await client.post(
MARKET_STOCK_MCP_URL,
json=request,
headers={
"Content-Type": "application/json"
}
)
post_end = time.time()
print(f"[TIMING] HTTP POST took: {post_end - post_start:.3f}s")
print(f"[TIMING] Response status: {response.status_code}")
print(f"[TIMING] Response size: {len(response.content)} bytes")
response.raise_for_status()
result = response.json()
http_end = time.time()
print(f"[TIMING] Total HTTP operation took: {http_end - http_start:.3f}s")
return result
# Run async function
async_start = time.time()
print(f"[TIMING] Starting async execution...")
try:
# Try to get the current event loop
try:
loop = asyncio.get_running_loop()
print(f"[TIMING] Detected running event loop, applying nest_asyncio...")
nest_start = time.time()
# If we're already in an async context, we need to use nest_asyncio
import nest_asyncio
nest_asyncio.apply()
nest_end = time.time()
print(f"[TIMING] nest_asyncio.apply() took: {nest_end - nest_start:.3f}s")
# Run the async function
run_start = time.time()
response = asyncio.run(make_request())
run_end = time.time()
print(f"[TIMING] asyncio.run() took: {run_end - run_start:.3f}s")
except RuntimeError:
# No event loop running, we can use asyncio.run
print(f"[TIMING] No running event loop, using asyncio.run()...")
run_start = time.time()
response = asyncio.run(make_request())
run_end = time.time()
print(f"[TIMING] asyncio.run() took: {run_end - run_start:.3f}s")
except Exception as async_error:
print(f"[DEBUG] Error in async execution: {str(async_error)}")
raise
async_end = time.time()
print(f"[TIMING] Total async operation took: {async_end - async_start:.3f}s")
# Check for error response
if "error" in response:
error_msg = f"MarketandStockMCP Error {response['error'].get('code', 'unknown')}: {response['error'].get('message', 'Unknown error')}"
print(f"[DEBUG] {error_msg}")
raise Exception(error_msg)
# Return the result
if "result" in response:
result = response["result"]
print(f"[DEBUG] MarketandStockMCP tool result: {result}")
total_time = time.time() - start_time
print(f"[TIMING] Total MarketandStockMCP call took: {total_time:.3f}s")
return result
else:
error_msg = "MarketandStockMCP tool call failed: no result in response"
print(f"[DEBUG] {error_msg}")
raise Exception(error_msg)
except httpx.HTTPStatusError as e:
error_msg = f"HTTP error calling MarketandStockMCP: {e.response.status_code} - {e.response.text}"
print(f"[DEBUG] HTTP status error: {error_msg}")
raise Exception(error_msg)
except httpx.RequestError as e:
error_msg = f"Request error calling MarketandStockMCP: {str(e)}"
print(f"[DEBUG] Request error: {error_msg}")
raise Exception(error_msg)
except json.JSONDecodeError as e:
error_msg = f"Failed to parse MarketandStockMCP response as JSON: {str(e)}"
print(f"[DEBUG] JSON decode error: {error_msg}")
raise Exception(error_msg)
except Exception as e:
# Ensure we have a meaningful error message
error_str = str(e) if str(e) else "Unknown error occurred"
output_messages.append(f"Error calling MarketandStockMCP tool {tool_name}: {error_str}")
print(f"[DEBUG] Exception in call_market_stock_mcp_tool: {error_str}")
print(f"[DEBUG] Exception type: {type(e)}")
# Print full traceback for debugging
import traceback
print(f"[DEBUG] Full traceback: {traceback.format_exc()}")
raise Exception(error_str)
def extract_url_from_user_input(user_input, hf_token):
"""Extract URL from user input using regex pattern matching"""
import re
# Use regular expressions to extract URLs directly from user input instead of relying on LLM
url_pattern = r'https?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+[/\w\-._~:/?#[\]@!$&\'()*+,;=%]*'
urls = re.findall(url_pattern, user_input)
if urls:
url = urls[0] # 返回找到的第一个URL
if url.startswith("http"):
return url
return None
def get_third_party_mcp_tools():
"""
Get information about third-party MCP tools (EasyReportsMCP)
"""
# Define the tools with their descriptions and schemas
tools = [
{
"name": "search_company",
"description": "Search for a company by name in SEC EDGAR database. Returns company CIK, name, and ticker symbol.",
"inputSchema": {
"type": "object",
"properties": {
"company_name": {
"type": "string",
"description": "Company name to search (e.g., 'Microsoft', 'Apple', 'Tesla')"
}
},
"required": ["company_name"]
}
},
{
"name": "get_company_info",
"description": "Get detailed company information including name, tickers, SIC code, and industry description.",
"inputSchema": {
"type": "object",
"properties": {
"cik": {
"type": "string",
"description": "Company CIK code (10-digit format, e.g., '0000789019')"
}
},
"required": ["cik"]
}
},
{
"name": "get_company_filings",
"description": "Get list of company SEC filings (10-K, 10-Q, 20-F, etc.) with filing dates and document links.",
"inputSchema": {
"type": "object",
"properties": {
"cik": {
"type": "string",
"description": "Company CIK code"
},
"form_types": {
"type": "array",
"items": {
"type": "string"
},
"description": "Optional: Filter by form types (e.g., ['10-K', '10-Q'])"
}
},
"required": ["cik"]
}
},
{
"name": "get_financial_data",
"description": "Get financial data for a specific period including revenue, net income, EPS, operating expenses, and cash flow.",
"inputSchema": {
"type": "object",
"properties": {
"cik": {
"type": "string",
"description": "Company CIK code"
},
"period": {
"type": "string",
"description": "Period in format 'YYYY' for annual or 'YYYYQX' for quarterly (e.g., '2024', '2024Q3')"
}
},
"required": ["cik", "period"]
}
},
{
"name": "extract_financial_metrics",
"description": "Extract comprehensive financial metrics for multiple years including both annual and quarterly data. Returns data in chronological order (newest first).",
"inputSchema": {
"type": "object",
"properties": {
"cik": {
"type": "string",
"description": "Company CIK code"
},
"years": {
"type": "integer",
"description": "Number of recent years to extract (1-10, default: 3)",
"minimum": 1,
"maximum": 10,
"default": 3
}
},
"required": ["cik"]
}
},
{
"name": "get_latest_financial_data",
"description": "Get the most recent financial data available for a company.",
"inputSchema": {
"type": "object",
"properties": {
"cik": {
"type": "string",
"description": "Company CIK code"
}
},
"required": ["cik"]
}
},
{
"name": "advanced_search_company",
"description": "Advanced search supporting both company name and CIK code. Automatically detects input type.",
"inputSchema": {
"type": "object",
"properties": {
"company_input": {
"type": "string",
"description": "Company name, ticker, or CIK code"
}
},
"required": ["company_input"]
}
}
]
return tools
def get_market_stock_mcp_tools():
"""
Get information about MarketandStockMCP tools
"""
tools = [
{
"name": "get_quote",
"description": "Get real-time quote data for US stocks. Use this when you need current stock price information and market performance metrics for any US-listed stock.",
"inputSchema": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Stock ticker symbol (e.g., 'AAPL', 'MSFT', 'TSLA', 'GOOGL')"
}
},
"required": ["symbol"]
}
},
{
"name": "get_market_news",
"description": "Get latest market news across different categories (general, forex, crypto, merger). Use this when you need current market news, trends, and developments.",
"inputSchema": {
"type": "object",
"properties": {
"category": {
"type": "string",
"enum": ["general", "forex", "crypto", "merger"],
"description": "News category: 'general' (market news), 'forex' (currency news), 'crypto' (cryptocurrency news), or 'merger' (M&A news)",
"default": "general"
},
"min_id": {
"type": "integer",
"description": "Minimum news ID for pagination (default: 0)",
"default": 0
}
},
"required": []
}
},
{
"name": "get_company_news",
"description": "Get latest news for a specific company by stock symbol. Only available for North American companies. Use this when you need company-specific announcements or press releases.",
"inputSchema": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Company stock ticker symbol (e.g., 'AAPL', 'MSFT', 'TSLA', 'GOOGL'). Must be a North American (US/Canada) listed company."
},
"from_date": {
"type": "string",
"description": "Start date in YYYY-MM-DD format (default: 7 days ago)"
},
"to_date": {
"type": "string",
"description": "End date in YYYY-MM-DD format (default: today)"
}
},
"required": ["symbol"]
}
}
]
return tools
def get_available_mcp_tools(mcp_process):
"""
Get information about all available MCP tools including third-party tools
"""
try:
# First ensure session is initialized
global MCP_INITIALIZED
if not MCP_INITIALIZED:
print("Initializing MCP session for tool discovery")
success = initialize_mcp_session_stdio(mcp_process)
if not success:
raise Exception("Failed to initialize MCP session")
MCP_INITIALIZED = True
print("MCP session initialized successfully for tool discovery")
# Send tools/list request (without params field as per MCP spec)
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list"
}
print(f"Sending tools/list request: {request}")
# Send request to MCP process via stdin
request_str = json.dumps(request) + "\n"
mcp_process.stdin.write(request_str.encode('utf-8'))
mcp_process.stdin.flush()
# Read response from MCP process via stdout
response_data = b""
timeout = time.time() + 10 # 10 second timeout
while time.time() < timeout:
# Read a byte at a time until we get a newline
byte = mcp_process.stdout.read(1)
if not byte:
# Check if process is still alive
if mcp_process.poll() is not None:
raise Exception("MCP process terminated unexpectedly during tools list")
time.sleep(0.01) # Small delay to prevent busy waiting
continue
response_data += byte
if byte == b'\n':
break
else:
raise Exception("Timeout waiting for MCP tools list response")
# Parse the response
response_str = response_data.decode('utf-8').strip()
if not response_str:
raise Exception("Empty response from MCP tools list")
print(f"MCP tools list response: {response_str}")
response = json.loads(response_str)
# Check for error response
if "error" in response:
error_msg = f"MCP Error {response['error'].get('code', 'unknown')}: {response['error'].get('message', 'Unknown error')}"
# Add additional context for common errors
if response['error'].get('code') == -32602:
error_msg += f". This typically means the request parameters are invalid. Request: {json.dumps(request)}"
raise Exception(error_msg)
# Get local MCP tools
local_tools = []
if "result" in response and "tools" in response["result"]:
local_tools = response["result"]["tools"]
else:
raise Exception("MCP tools list failed: no tools in response")
# Get third-party MCP tools (EasyReportsMCP)
third_party_tools = get_third_party_mcp_tools()
# ✅ Get MarketandStockMCP tools
market_stock_tools = get_market_stock_mcp_tools()
# Combine all tool lists
all_tools = local_tools + third_party_tools + market_stock_tools
print(f"Combined tools: {len(local_tools)} local + {len(third_party_tools)} third-party + {len(market_stock_tools)} market/stock = {len(all_tools)} total")
return all_tools
except json.JSONDecodeError as e:
raise Exception(f"Failed to parse MCP tools list response as JSON: {str(e)}")
except Exception as e:
print(f"Error getting MCP tools list: {str(e)}")
raise e
def decide_tool_execution_plan(user_message, tools_info, history, hf_token, agent_context=None):
"""
Let LLM decide which tools to use based on user request
Args:
user_message: User's request
tools_info: List of available tools
history: Conversation history
hf_token: HuggingFace token
agent_context: Agent context with previously stored information (optional)
"""
# DEBUG: Print the actual user message
print(f"[DEBUG] User message received in decide_tool_execution_plan: '{user_message}'")
print(f"[DEBUG] User message type: {type(user_message)}")
print(f"[DEBUG] User message length: {len(user_message) if user_message else 0}")
try:
client = InferenceClient(
model="Qwen/Qwen2.5-72B-Instruct",
token=hf_token if hf_token else None
)
# Format tools information for LLM
tools_description = "\n".join([
f"Tool: {tool['name']}\nDescription: {repr(tool['description'].strip())}\nParameters: {json.dumps(tool.get('inputSchema', {}), indent=2)}"
for tool in tools_info
])
# Format conversation history for LLM context
history_context = ""
if history:
history_context = "\nPrevious conversation context:\n"
for i, (user_msg, assistant_msg) in enumerate(history[-3:]): # Include last 3 exchanges
history_context += f"User: {user_msg}\nAssistant: {assistant_msg}\n"
# Format agent context for LLM
context_info = ""
if agent_context and len(agent_context) > 0:
context_info = "\n\nAgent Context (previously gathered information that can be reused):\n"
if 'last_company_name' in agent_context:
context_info += f"- Last company: {agent_context['last_company_name']}"
if 'last_company_ticker' in agent_context:
context_info += f" ({agent_context['last_company_ticker']})"
context_info += "\n"
if 'last_company_cik' in agent_context:
context_info += f"- Company CIK: {agent_context['last_company_cik']}\n"
if 'last_period' in agent_context:
context_info += f"- Last period: {agent_context['last_period']}\n"
if 'last_financial_report_url' in agent_context:
context_info += f"- Last report URL: {agent_context['last_financial_report_url']}\n"
# Include financial data summary if available
if 'last_financial_data' in agent_context:
data = agent_context['last_financial_data']
context_info += "- Last financial data available:\n"
if 'total_revenue' in data:
context_info += f" Revenue: ${data['total_revenue']:,}\n"
if 'net_income' in data:
context_info += f" Net Income: ${data['net_income']:,}\n"
if 'earnings_per_share' in data:
context_info += f" EPS: ${data['earnings_per_share']}\n"
context_info += "\n**CRITICAL CONTEXT USAGE RULES:**\n"
context_info += "1. If the user is asking follow-up questions about the SAME company, you can skip search_company and directly use the CIK from context.\n"
context_info += "2. **If the user asks to 'analyze this report' or 'analyze the financial report' and last_financial_data is available, return an EMPTY tool plan [] - the system will use the context data directly for analysis.**\n"
context_info += "3. **DO NOT call analyze_financial_report_file for follow-up analysis requests when financial data is already in context.**\n"
context_info += "4. Only call new tools if the user is asking for DIFFERENT data (different company, different period, etc.).\n"
# Create prompt for LLM to decide tool execution plan
prompt = f"""
You are a financial analysis assistant that can use various tools to help users.
**USER'S ACTUAL REQUEST: {user_message}**
Available tools:
{tools_description}
{context_info}
{history_context}
Based on the user's request above, decide which tools to use and in what order.
Provide your response in the following JSON format:
{{
"plan": [
{{
"tool": "tool_name",
"arguments": {{
"param1": "value1",
"param2": "value2"
}},
"reason": "reason for using this tool"
}}
],
"explanation": "brief explanation of your plan"
}}
Important guidelines:
1. If the user mentions a company name but no direct URL, you should first try to extract a valid URL from their message
2. If no valid URL is found, and the user is asking for analysis of a specific company's financial report, use the search_and_extract_financial_report tool
3. If the search_and_extract_financial_report tool is used and returns guidance, present that guidance to the user
4. If the search_and_extract_financial_report tool returns URLs, you should analyze the search results to select the most appropriate URL for financial analysis
5. For URLs returned by search tools or provided by users that are not PDF files, you can analyze them directly without downloading
6. Always validate URLs before using the download_financial_report tool
7. If the user provides an invalid URL, suggest alternatives or ask for a working one
8. ONLY include tools in the plan that you are certain can and should be executed
9. If you cannot determine a valid plan, return an empty plan array
10. For a complete financial analysis workflow, you typically need to:
- First search for financial reports (search_and_extract_financial_report) if no URL is provided
- Or download the financial report (download_financial_report) if a URL is provided
- Then analyze the downloaded report file directly (analyze_financial_report_file)
11. Plan all necessary steps in the correct order based on the user's request
12. If the user is asking follow-up questions about a previous analysis (like "should I buy or sell?"), and there was a recent successful analysis, you can provide insights based on that context
13. If the user requests analysis for a company by name (e.g., "analyze Amazon's financial report") and no URL is available:
- First, try to use the search_and_extract_financial_report tool to help find the report
- If that tool provides guidance, present it to the user with clear next steps
- Explain that you can analyze financial reports once a valid source is provided
14. When searching for financial reports, prioritize the most recent reports to enable trend analysis
15. When analyzing financial data, always look for trends over multiple periods and compare current performance with historical data
16. When asking users for financial report URLs, use the term "URL (or PDF format URL)" to indicate that both regular web URLs and PDF URLs are acceptable
17. When the search_and_extract_financial_report tool returns search results, you should carefully analyze them to select the most appropriate URL for financial analysis. Consider the following factors:
- Prefer PDF files over web pages for more reliable analysis
- Look for official sources (company website, SEC.gov)
- Prioritize recent annual reports (10-K) over quarterly reports (10-Q)
- Choose reports with comprehensive financial statements
- Look for URLs containing keywords like "10-K", "annual-report", "financial-statement"
- Select the most recent reports when multiple options are available
18. After analyzing the search results, you should explicitly choose the best URL and use the analyze_financial_report_file tool to analyze it
19. You have full autonomy to construct search terms based on user intent and analyze search results to fulfill user requests
29. Only use financial analysis tools when you are certain the user wants detailed analysis of a specific company's financial reports
30. When in doubt, engage in natural conversation and ask the user if they would like to proceed with detailed financial analysis
31. Never assume the user wants financial report analysis just because they mentioned a company name
32. Always confirm with the user before proceeding with detailed financial analysis tools
33. If search results are empty or contain no relevant financial reports, gracefully return to natural conversation without attempting to force analysis
34. Avoid attempting to analyze empty or irrelevant search results - this will lead to poor user experience
35. When search results are unhelpful, acknowledge this and continue with normal conversation flow
36. For general inquiries or conversational requests that don't require financial analysis tools, return an empty tool plan and engage in natural conversation
37. Only initiate the financial report processing service when you have determined that specific financial analysis tools are needed
38. Avoid starting financial analysis workflows for general questions, advice requests, or conversational topics
39. When the user requests financial report search, pass the complete user query as the "user_query" parameter to the search_and_extract_financial_report tool
40. Example of correct parameter format for search_and_extract_financial_report tool:
{{
"tool": "search_and_extract_financial_report",
"arguments": {{
"user_query": "Apple Inc. annual report 2024 PDF"
}},
"reason": "User requested financial analysis for Apple Inc."
}}
41. If the user is asking for a specific financial report download link (e.g., "What is the download URL for Alibaba FY 2025 Annual Report?"), you should:
- First use the search_and_extract_financial_report tool to find relevant search results
- Then use the deep_analyze_and_extract_download_link tool to analyze the search results and extract the most relevant download link
- Present the extracted download link to the user without initiating full financial analysis
42. The deep_analyze_and_extract_download_link tool should be used when you need to find specific download links from search results rather than performing full financial analysis
43. Example of correct parameter format for deep_analyze_and_extract_download_link tool:
{{
"tool": "deep_analyze_and_extract_download_link",
"arguments": {{
"search_results": ["search results array from previous tool"],
"user_request": "User's specific request for download link"
}},
"reason": "User requested a specific download link for a financial report"
}}
44. **CRITICAL**: When using analyze_financial_report_file, DO NOT make up or guess the filename parameter. Instead:
- If you don't know the actual filename, set it to null or an empty string: "filename": ""
- The system will automatically find the most recently downloaded file
- Never use placeholder names like "Microsoft_FY25_Q1_Report.pdf" or "report.pdf"
- Only provide a specific filename if the user explicitly mentioned it or it was returned by a previous tool
45. **CRITICAL**: For financial data queries, be selective with tool usage:
- Prioritize using third-party MCP tools (SEC EDGAR data) as they provide authoritative source data
- For a simple query like "[Company] 2025 Q1 financial report", you typically need:
1. search_company (to get the company's CIK code)
2. get_financial_data (to get specific period financial data)
- AVOID using get_company_filings unless the user specifically asks for filing lists
- AVOID using extract_financial_metrics unless the user asks for multi-year analysis
- Use the minimum number of tools necessary to answer the user's question
- Each tool has overhead - be efficient and focused
- **CRITICAL: Always extract the company name from the USER'S ACTUAL REQUEST - never use examples or made-up company names!**
46. Example of correct usage when filename is unknown:
{{
"tool": "analyze_financial_report_file",
"arguments": {{
"filename": "" // Empty string - system will auto-fill with latest downloaded file
}},
"reason": "Analyze the previously downloaded financial report"
}}
If no tools are needed, return an empty plan array.
"""
messages = [
{"role": "system", "content": "You are a precise JSON generator that helps decide which tools to use for financial analysis. Always plan the minimum necessary tools to answer the user's question efficiently. For financial data queries, typically use search_company + get_financial_data. Avoid extra tools unless explicitly needed. CRITICAL: Always extract company names and other parameters from the USER'S ACTUAL REQUEST - never use example data or make up information. IMPORTANT: Output ONLY valid JSON without any comments or explanations."},
{"role": "user", "content": prompt}
]
# Get response from LLM
response = client.chat.completions.create(
model="Qwen/Qwen2.5-72B-Instruct",
messages=messages,
max_tokens=500,
temperature=0.3,
)
# Extract the JSON response
if hasattr(response, 'choices') and len(response.choices) > 0:
content = response.choices[0].message.content if hasattr(response.choices[0].message, 'content') else str(response.choices[0].message)
else:
content = str(response)
# Debug: Log the raw LLM response
print(f"[DEBUG] Raw LLM response for tool planning: {content}")
# Try to parse as JSON
try:
# Extract JSON from the response if it's wrapped in other text
import re
json_match = re.search(r'\{.*\}', content, re.DOTALL)
if json_match:
json_str = json_match.group(0)
# Remove JavaScript-style comments (// ...) from JSON
# This is a common issue where LLMs add explanatory comments in JSON
lines = json_str.split('\n')
cleaned_lines = []
for line in lines:
# Remove // comments but keep the rest of the line
if '//' in line:
# Find the position of //
comment_pos = line.find('//')
# Check if // is inside a string
before_comment = line[:comment_pos]
# Count quotes before comment to see if we're inside a string
quote_count = before_comment.count('"') - before_comment.count('\\"')
if quote_count % 2 == 0:
# Even number of quotes = we're outside a string, safe to remove comment
line = line[:comment_pos].rstrip()
# Also remove trailing comma if present
if line.rstrip().endswith(','):
line = line.rstrip()[:-1].rstrip()
cleaned_lines.append(line)
json_str = '\n'.join(cleaned_lines)
result = json.loads(json_str)
# Ensure plan is empty if no valid plan can be made
if not result.get("plan"):
result["plan"] = []
return result
else:
# If no JSON found, return empty plan
return {"plan": [], "explanation": content if content else "No valid plan could be generated"}
except json.JSONDecodeError as e:
# If JSON parsing fails, return a default plan
print(f"Failed to parse LLM response as JSON: {content}")
print(f"JSON decode error: {str(e)}")
return {"plan": [], "explanation": "Could not generate a tool execution plan"}
except Exception as e:
error_msg = str(e)
print(f"Error in decide_tool_execution_plan: {error_msg}")
# Check if it's a 5xx server error (retryable)
if "500" in error_msg or "502" in error_msg or "503" in error_msg or "504" in error_msg:
return {"plan": [], "explanation": "API temporarily unavailable. Please try again in a moment."}
else:
return {"plan": [], "explanation": f"Error generating plan: {error_msg}"}
def execute_tool_plan(mcp_process, tool_plan, output_messages, user_message, hf_token=None):
"""
Execute the tool plan generated by LLM
"""
results = []
successful_tools = 0 # Track number of successful tools
search_returned_no_results = False # Initialize search result flag
try:
# Keep track of downloaded filename for use in analyze_financial_report_file
downloaded_filename = None
for step in tool_plan.get("plan", []):
tool_name = step.get("tool")
arguments = step.get("arguments", {})
reason = step.get("reason", "")
# Log the tool execution plan for debugging
output_messages.append(f"🔧 Tool execution plan - Tool: {tool_name}, Arguments: {json.dumps(arguments, indent=2, ensure_ascii=False)}")
yield "\n".join(output_messages)
output_messages.append("") # Add empty line for spacing
yield "\n".join(output_messages)
# Special handling for download_financial_report tool
if tool_name == "download_financial_report":
url = arguments.get("url", "")
# If no URL provided, skip this tool
if not url:
output_messages.append("⚠️ Skipping download_financial_report tool: No URL provided")
yield "\n".join(output_messages)
continue
output_messages.append(f"🔍 Validating URL: {url}")
yield "\n".join(output_messages)
# Validate URL before attempting download
if not validate_url(url):
output_messages.append(f"⚠️ The URL {url} appears to be invalid or inaccessible.")
output_messages.append("💡 Please provide a valid financial report URL (or PDF format URL) for analysis, for example:")
output_messages.append(" • https://somecompany.com/reports/annual-report-2024.pdf")
output_messages.append(" • https://investors.somecompany.com/financials/2024-q3-report.pdf")
output_messages.append(" • https://somecompany.com/investor-relations/financial-reports/")
yield "\n".join(output_messages)
# Skip this tool and continue with others
continue
# Special handling for search_and_extract_financial_report tool
elif tool_name == "search_and_extract_financial_report":
user_query = arguments.get("user_query", "")
if not user_query:
output_messages.append("⚠️ Skipping search_and_extract_financial_report tool: No user query provided")
yield "\n".join(output_messages)
continue
# Special handling for analyze_financial_report_file tool - auto-fill filename if not provided
elif tool_name == "analyze_financial_report_file" and not arguments.get("filename"):
if downloaded_filename:
# Auto-fill the filename from the previous download
arguments = arguments.copy() # Create a copy to avoid modifying the original
arguments["filename"] = downloaded_filename
output_messages.append(f"🔧 Auto-filling filename for analyze_financial_report_file: {downloaded_filename}")
yield "\n".join(output_messages)
else:
# Try to get the most recent downloaded file
try:
list_result = call_mcp_tool_stdio(mcp_process, "list_downloaded_reports", {})
if list_result and "reports" in list_result and list_result["reports"]:
# Get the most recently modified file
reports = sorted(list_result["reports"], key=lambda x: x.get("modified", ""), reverse=True)
if reports:
downloaded_filename = reports[0]["filename"]
arguments = arguments.copy()
arguments["filename"] = downloaded_filename
output_messages.append(f"🔧 Auto-filling filename for analyze_financial_report_file: {downloaded_filename}")
yield "\n".join(output_messages)
except Exception as e:
output_messages.append(f"⚠️ Could not auto-fill filename for analyze_financial_report_file: {str(e)}")
yield "\n".join(output_messages)
# CRITICAL: Auto-fill source_url parameter if available from download
if tool_name == "analyze_financial_report_file" and "source_url" not in arguments:
# Look for source_url from the most recent download
for prev_result in reversed(results):
if prev_result.get("tool") == "download_financial_report":
tool_result = prev_result.get("result")
if tool_result and isinstance(tool_result, dict) and "source_url" in tool_result:
arguments = arguments.copy()
arguments["source_url"] = tool_result["source_url"]
output_messages.append(f"🔗 Including source URL for analysis context")
yield "\n".join(output_messages)
break
# CRITICAL: Auto-fill CIK parameter for get_financial_data from search_company result
if tool_name == "get_financial_data" and "cik" in arguments:
# Check if there was a recent search_company call
for prev_result in reversed(results):
if prev_result.get("tool") == "search_company":
tool_result = prev_result.get("result")
if tool_result:
# Try to extract CIK from MCP content format
try:
if 'content' in tool_result and isinstance(tool_result['content'], list):
for content_item in tool_result['content']:
if isinstance(content_item, dict) and 'text' in content_item:
parsed_data = json.loads(content_item['text'])
if isinstance(parsed_data, dict) and 'cik' in parsed_data:
correct_cik = parsed_data['cik']
if arguments['cik'] != correct_cik:
output_messages.append(f"🔧 Correcting CIK: {arguments['cik']} → {correct_cik}")
arguments = arguments.copy()
arguments['cik'] = correct_cik
yield "\n".join(output_messages)
break
except (json.JSONDecodeError, KeyError) as e:
print(f"[DEBUG] Could not extract CIK from search_company result: {e}")
break
output_messages.append(f"🤖 Agent Decision: {reason}")
yield "\n".join(output_messages)
output_messages.append(f"🔧 Agent Action: Calling tool '{tool_name}' with arguments {json.dumps(arguments, indent=2, ensure_ascii=False)}")
yield "\n".join(output_messages)
# CRITICAL: Resolve argument references before calling the tool
# If arguments contain references to previous tool outputs, replace them with actual data
# print(f"[DEBUG] Tool arguments before resolution: {json.dumps(arguments, indent=2)}")
if isinstance(arguments, dict):
for arg_name, arg_value in arguments.items():
print(f"[DEBUG] Checking argument '{arg_name}' = '{arg_value}' (type: {type(arg_value)})")
# Detect if this looks like a reference to previous tool output
# Strategy: If it's a string that should be a list/dict (like search_results),
# and the string value is just a simple identifier or reference phrase,
# try to resolve it from previous tool results
is_reference = False
if isinstance(arg_value, str):
# Heuristic: If the argument name suggests it should be structured data (results, data, list, etc.)
# but the value is a simple string (no spaces, or looks like a reference),
# it's likely a reference that needs resolution
arg_name_lower = arg_name.lower()
arg_value_lower = arg_value.lower()
# Check if argument name suggests it should be structured data
expects_structured_data = any(keyword in arg_name_lower for keyword in [
'results', 'data', 'list', 'items', 'entries', 'records'
])
# Check if value looks like a reference (not actual data)
looks_like_reference = (
# Simple identifier matching the parameter name
arg_value_lower == arg_name_lower or
# Contains reference keywords
any(keyword in arg_value_lower for keyword in [
'previous', 'from', 'result', 'output', 'tool'
]) or
# Has dot notation (like "tool.results")
'.' in arg_value_lower
)
if expects_structured_data and looks_like_reference:
is_reference = True
if is_reference:
print(f"[DEBUG] Detected argument reference: {arg_value}")
# Try to find and resolve the reference from previous tool results
actual_data = None
# Look through all previous tool results (most recent first)
for prev_result in reversed(results):
prev_tool = prev_result.get("tool")
tool_result = prev_result.get("result")
if not tool_result:
continue
# Try to extract structured data from the tool result
# Handle MCP 'content' array format
if 'content' in tool_result and isinstance(tool_result['content'], list):
for content_item in tool_result['content']:
if isinstance(content_item, dict) and 'text' in content_item:
try:
parsed_data = json.loads(content_item['text'])
# Look for fields that match what we need
if isinstance(parsed_data, dict):
# Try to find a field that looks like it contains the data
for key in ['results', 'data', 'items', 'entries']:
if key in parsed_data and isinstance(parsed_data[key], list):
actual_data = parsed_data[key]
print(f"[DEBUG] Resolved reference from tool '{prev_tool}' field '{key}': {len(actual_data)} items")
break
# If we're looking for a simple value like CIK, check for direct fields
if actual_data is None and arg_name in parsed_data:
actual_data = parsed_data[arg_name]
print(f"[DEBUG] Resolved simple value reference from tool '{prev_tool}' field '{arg_name}': {actual_data}")
if actual_data:
break
except json.JSONDecodeError:
continue
# Handle structuredContent format
if not actual_data and 'structuredContent' in tool_result:
struct_content = tool_result['structuredContent']
if isinstance(struct_content, dict) and 'result' in struct_content:
result_data = struct_content['result']
if isinstance(result_data, dict):
for key in ['results', 'data', 'items', 'entries']:
if key in result_data and isinstance(result_data[key], list):
actual_data = result_data[key]
print(f"[DEBUG] Resolved reference from tool '{prev_tool}' structuredContent: {len(actual_data)} items")
break
# If we're looking for a simple value like CIK, check for direct fields
if actual_data is None and arg_name in result_data:
actual_data = result_data[arg_name]
print(f"[DEBUG] Resolved simple value reference from tool '{prev_tool}' structuredContent field '{arg_name}': {actual_data}")
# Handle direct result format (for simple values like CIK)
if not actual_data and isinstance(tool_result, dict):
# Look for the argument name directly in the result
if arg_name in tool_result:
actual_data = tool_result[arg_name]
print(f"[DEBUG] Resolved simple value reference from tool '{prev_tool}' direct field '{arg_name}': {actual_data}")
# Also check common fields that might contain the data
elif 'result' in tool_result and isinstance(tool_result['result'], dict):
if arg_name in tool_result['result']:
actual_data = tool_result['result'][arg_name]
print(f"[DEBUG] Resolved simple value reference from tool '{prev_tool}' result field '{arg_name}': {actual_data}")
elif 'content' in tool_result and isinstance(tool_result['content'], list) and len(tool_result['content']) > 0:
# Check if content contains a simple text result
content_item = tool_result['content'][0]
if isinstance(content_item, dict) and 'text' in content_item:
try:
# Try to parse as JSON first
parsed_text = json.loads(content_item['text'])
if isinstance(parsed_text, dict) and arg_name in parsed_text:
actual_data = parsed_text[arg_name]
print(f"[DEBUG] Resolved simple value reference from tool '{prev_tool}' content JSON field '{arg_name}': {actual_data}")
except json.JSONDecodeError:
# If not JSON, check if it's the actual value we're looking for
if arg_name == 'cik' and content_item['text'].startswith('{'):
# It might be JSON in string format
try:
parsed_json = json.loads(content_item['text'])
if isinstance(parsed_json, dict) and 'cik' in parsed_json:
actual_data = parsed_json['cik']
print(f"[DEBUG] Resolved CIK from JSON content: {actual_data}")
except:
pass
if actual_data:
break
if actual_data is not None:
# Replace the reference with actual data
arguments[arg_name] = actual_data
print(f"[DEBUG] Replaced argument '{arg_name}' with actual data ({len(actual_data)} items)")
else:
print(f"[DEBUG] WARNING: Could not resolve reference '{arg_value}' - no suitable data found in previous results")
# Call the tool
try:
# CRITICAL: Check which MCP service to use
is_third_party_tool = tool_name in THIRD_PARTY_MCP_TOOLS
is_market_stock_tool = tool_name in MARKET_STOCK_MCP_TOOLS
if is_third_party_tool:
# Add progress message for potentially slow operations
if tool_name == "get_financial_data":
output_messages.append("🔍 Fetching detailed financial data from SEC EDGAR... (this may take 30-60 seconds)")
yield "\n".join(output_messages)
elif tool_name in ["get_company_filings", "extract_financial_metrics"]:
output_messages.append(f"🔍 Retrieving data from SEC database... (this may take a moment)")
yield "\n".join(output_messages)
# Call third-party MCP tool via HTTP
tool_result = call_third_party_mcp_tool(tool_name, arguments)
elif is_market_stock_tool:
# ✅ Add progress message for MarketandStockMCP operations
if tool_name == "get_quote":
output_messages.append("💹 Fetching real-time stock quote...")
yield "\n".join(output_messages)
elif tool_name == "get_market_news":
output_messages.append("📰 Retrieving latest market news...")
yield "\n".join(output_messages)
elif tool_name == "get_company_news":
output_messages.append("📰 Fetching company-specific news...")
yield "\n".join(output_messages)
# ✅ Call MarketandStockMCP tool via HTTP
tool_result = call_market_stock_mcp_tool(tool_name, arguments)
else:
# Call local MCP tool via stdio
tool_result = call_mcp_tool_stdio(mcp_process, tool_name, arguments)
results.append({
"tool": tool_name,
"arguments": arguments,
"result": tool_result,
"success": True # Mark as successful
})
successful_tools += 1 # Increment successful tools counter
# If this was a download tool, save the filename for later use
if tool_name == "download_financial_report" and tool_result and "filename" in tool_result:
downloaded_filename = tool_result["filename"]
output_messages.append(f"📎 Downloaded file: {downloaded_filename}")
# CRITICAL: Update session URL when new download happens
if "source_url" in tool_result:
current_session_url = tool_result["source_url"]
print(f"[SESSION] Updated session URL: {current_session_url}")
# If this was a search tool that returned guidance, present it to the user
if tool_name == "search_and_extract_financial_report" and tool_result and tool_result.get("type") == "search_guidance":
guidance_message = tool_result.get("message", "")
suggestion = tool_result.get("suggestion", "")
output_messages.append(f"💡 {guidance_message}")
if suggestion:
output_messages.append(f"📋 {suggestion}")
yield "\n".join(output_messages)
# If this was a search tool that found real results, present them to the user and prepare for direct analysis
elif tool_name == "search_and_extract_financial_report" and tool_result:
# First, extract the actual result from the tool_result structure
actual_result = None
if isinstance(tool_result, dict):
# Handle structuredContent format (newer MCP responses)
if "structuredContent" in tool_result and "result" in tool_result["structuredContent"]:
actual_result = tool_result["structuredContent"]["result"]
print(f"[DEBUG] Using structuredContent result format")
# Handle direct result format (older MCP responses)
elif "result" in tool_result:
actual_result = tool_result["result"]
print(f"[DEBUG] Using direct result format")
# Handle direct format (when result is directly in tool_result)
else:
actual_result = tool_result
print(f"[DEBUG] Using tool_result directly")
else:
actual_result = tool_result
print(f"[DEBUG] tool_result is not a dict, using directly")
print(f"[DEBUG] actual_result type: {type(actual_result)}")
if isinstance(actual_result, dict):
print(f"[DEBUG] actual_result keys: {list(actual_result.keys())}")
# Check if this is a search results response
if actual_result.get("type") == "search_results":
search_message = actual_result.get("message", "")
output_messages.append(f"🔍 {search_message}")
yield "\n".join(output_messages)
# Debug: Print the structure of actual_result
# print(f"[DEBUG] search_and_extract_financial_report actual_result: {json.dumps(actual_result, indent=2)[:500]}...")
# Extract URLs from the search results for potential direct analysis
print(f"[DEBUG] actual_result type: {type(actual_result)}")
print(f"[DEBUG] actual_result keys: {list(actual_result.keys()) if isinstance(actual_result, dict) else 'Not a dict'}")
links = actual_result.get("results", [])
print(f"[DEBUG] links type: {type(links)}")
print(f"[DEBUG] links length: {len(links)}")
if len(links) > 0:
print(f"[DEBUG] first link type: {type(links[0])}")
print(f"[DEBUG] first link keys: {list(links[0].keys()) if isinstance(links[0], dict) else 'Not a dict'}")
if links:
# Display all search results to the user for Agent analysis
output_messages.append("📋 Search Results:")
# Show all results, not just top 5, to give Agent more options
for i, link in enumerate(links, 1):
title = link.get("title", "No Title")
url = link.get("link", "No URL")
snippet = link.get("snippet", "")
output_messages.append(f"{i}. {title}")
output_messages.append(f" URL: {url}")
if snippet:
output_messages.append(f" Summary: {snippet}")
output_messages.append("")
print(f"[DEBUG] About to yield search results to user")
try:
yield "\n".join(output_messages)
print(f"[DEBUG] Search results displayed to user")
except Exception as e:
print(f"[ERROR] Failed to yield search results: {str(e)}")
import traceback
traceback.print_exc()
# Check if user is requesting download links (not full analysis)
# Keywords that indicate user wants download links
download_link_keywords = [
"download link", "download url", "下载链接", "链接",
"pdf link", "report link", "where to download",
"how to download", "link to", "url for"
]
user_wants_download_link = any(
keyword in user_message.lower()
for keyword in download_link_keywords
)
# If user wants download links, automatically call deep_analyze_and_extract_download_link
# But ONLY if it's not already in the tool plan to avoid duplication
should_auto_extract = user_wants_download_link
# Check if deep_analyze_and_extract_download_link is already in the plan
for remaining_step in tool_plan.get("plan", []):
if remaining_step.get("tool") == "deep_analyze_and_extract_download_link":
should_auto_extract = False
print(f"[DEBUG] Skipping auto-extraction because deep_analyze_and_extract_download_link is already in the plan")
break
if should_auto_extract:
output_messages.append("🧠 Detected that you want download links. Analyzing search results to extract the best download link...")
yield "\n".join(output_messages)
try:
# Call deep_analyze_and_extract_download_link tool
deep_analysis_result = call_mcp_tool_stdio(
mcp_process,
"deep_analyze_and_extract_download_link",
{
"search_results": links,
"user_request": user_message
}
)
# Add this result to the results list so it can be processed later
results.append({
"tool": "deep_analyze_and_extract_download_link",
"arguments": {
"search_results": links,
"user_request": user_message
},
"result": deep_analysis_result,
"success": True # Mark as successful
})
successful_tools += 1
# Don't display results here - let the final Agent Response handle it
# This avoids duplicate display
except Exception as e:
print(f"[ERROR] Failed to call deep_analyze_and_extract_download_link: {str(e)}")
import traceback
traceback.print_exc()
# Continue with normal flow if deep analysis fails
output_messages.append("⚠️ Could not automatically extract download links. Showing search results instead.")
yield "\n".join(output_messages)
else:
# User wants full analysis, not just download links
# Instead of automatically analyzing the first URL,
# let the Agent decide which URL to analyze based on the search results
output_messages.append("🧠 Please analyze the search results above and decide which financial report URL to analyze.")
output_messages.append("💡 Consider factors like:")
output_messages.append(" • Prefer PDF files over web pages")
output_messages.append(" • Look for official sources (company website, SEC.gov)")
output_messages.append(" • Prioritize recent annual reports (10-K) over quarterly reports (10-Q)")
output_messages.append(" • Choose reports with comprehensive financial statements")
output_messages.append(" • Select the most recent reports when multiple options are available")
output_messages.append("")
output_messages.append("Please select the most suitable URL from the search results and then use the analyze_financial_report_file tool to analyze it.")
try:
yield "\n".join(output_messages)
except Exception as e:
print(f"[ERROR] Failed to yield analysis guidance: {str(e)}")
import traceback
traceback.print_exc()
# If this was a search tool that found no results, present the guidance
elif actual_result.get("type") == "search_no_results":
no_results_message = actual_result.get("message", "")
suggestion = actual_result.get("suggestion", "")
output_messages.append(f"⚠️ {no_results_message}")
if suggestion:
output_messages.append(f"📋 {suggestion}")
yield "\n".join(output_messages)
# If search returned no results, don't force financial report analysis, but engage in natural conversation instead
# Set flag to skip subsequent tool analysis steps
search_returned_no_results = True
# If this was a search tool that had network errors, present the guidance
elif actual_result.get("type") == "search_error":
network_error_message = actual_result.get("message", "")
suggestion = actual_result.get("suggestion", "")
output_messages.append(f"❌ {network_error_message}")
if suggestion:
output_messages.append(f"📋 {suggestion}")
yield "\n".join(output_messages)
# If this was a search tool that had an exception, present the guidance
elif actual_result.get("type") == "search_exception":
exception_message = actual_result.get("message", "")
suggestion = actual_result.get("suggestion", "")
output_messages.append(f"❌ {exception_message}")
if suggestion:
output_messages.append(f"📋 {suggestion}")
yield "\n".join(output_messages)
# Handle cases where deep analysis found no results
elif actual_result.get("type") == "no_results":
no_results_message = actual_result.get("message", "")
suggestion = actual_result.get("suggestion", "")
output_messages.append(f"⚠️ {no_results_message}")
if suggestion:
output_messages.append(f"📋 {suggestion}")
yield "\n".join(output_messages)
# Handle cases where deep analysis encountered errors
elif actual_result.get("type") == "analysis_error":
error_message = actual_result.get("message", "")
suggestion = actual_result.get("suggestion", "")
output_messages.append(f"❌ {error_message}")
if suggestion:
output_messages.append(f"📋 {suggestion}")
yield "\n".join(output_messages)
except Exception as e:
error_msg = str(e)
# Check if this is a network-related error
if "network" in error_msg.lower() or "connect" in error_msg.lower() or "ssl" in error_msg.lower() or "timeout" in error_msg.lower():
# Special handling for analyze_financial_report_file tool
if tool_name == "analyze_financial_report_file":
output_messages.append(f"⚠️ Tool '{tool_name}' failed due to network issues. This may be due to network restrictions in the execution environment.")
output_messages.append("💡 You can try one of the following solutions:")
output_messages.append(" 1. Try again later when network conditions improve")
output_messages.append(" 2. Use a direct PDF URL that's more accessible")
output_messages.append(" 3. Download the PDF manually and upload it directly to the system")
else:
output_messages.append(f"⚠️ Tool '{tool_name}' failed due to network issues. This may be due to network restrictions in the execution environment. Please try again later or use a direct PDF URL.")
else:
output_messages.append(f"❌ Error executing tool '{tool_name}': {error_msg}")
# Add the failed tool to results with success=False
results.append({
"tool": tool_name,
"arguments": arguments,
"result": {"error": error_msg},
"success": False # Mark as failed
})
yield "\n".join(output_messages)
# Continue with other tools rather than failing completely
continue
# Add successful_tools count and search result flag to the results
results.append({
"successful_tools": successful_tools,
"search_returned_no_results": search_returned_no_results
})
yield results
except Exception as e:
output_messages.append(f"❌ Error executing tool plan: {str(e)}")
yield "\n".join(output_messages)
raise
def respond(
message,
history: list[tuple[str, str]],
session_url: str = "", # CRITICAL: 会话URL状态参数
agent_context: dict = None, # CRITICAL: Agent上下文（从上一轮传入）
):
"""
Main response function that integrates with MCP service
"""
# Use default values for removed UI parameters
system_message = "You are a financial analysis assistant. Provide concise investment insights from company financial reports."
max_tokens = 1024
temperature = 0.7
top_p = 0.95
# Initialize agent_context if None
if agent_context is None:
agent_context = {}
else:
# Make a copy to avoid modifying the input dict
agent_context = dict(agent_context)
# Get HF token from environment variables
hf_token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")
# DEBUG: Check token availability
if hf_token:
print(f"[AUTH] HF token found: {len(hf_token)} characters")
else:
print(f"[AUTH] ⚠️ WARNING: No HF token found in environment variables!")
print(f"[AUTH] Checked: HF_TOKEN and HUGGING_FACE_HUB_TOKEN")
global MCP_INITIALIZED
print(f"\n[SESSION] Starting new turn with session_url: {session_url}")
# CRITICAL: Initialize agent context if not provided
if agent_context is None:
agent_context = {}
# Log existing context
if agent_context:
print(f"[CONTEXT] Existing agent context: {list(agent_context.keys())}")
# CRITICAL: Track the current session's source URL across multiple turns
current_session_url = session_url # Start with previous session URL
# Collect all output messages
output_messages = []
# First, do a quick check if user message seems to require tools
# This avoids unnecessary service startup for basic conversations
try:
client = InferenceClient(
model="Qwen/Qwen2.5-72B-Instruct",
token=hf_token if hf_token else None
)
# Quick intent check
intent_check_prompt = f"""
Analyze the user's message and determine if they need financial analysis tools or just want to have a conversation.
User message: {message}
Respond with ONLY one word:
- "TOOLS" if the user is asking for financial report analysis, searching for financial reports, or downloading financial data
- "CONVERSATION" if the user is just greeting, asking general questions, or having a casual conversation
Response:"""
intent_response = client.chat.completions.create(
model="Qwen/Qwen2.5-72B-Instruct",
messages=[{"role": "user", "content": intent_check_prompt}],
max_tokens=10,
temperature=0.1,
)
intent = "CONVERSATION"
if hasattr(intent_response, 'choices') and len(intent_response.choices) > 0:
content = intent_response.choices[0].message.content if hasattr(intent_response.choices[0].message, 'content') else str(intent_response.choices[0].message)
if content:
intent = content.strip().upper()
# If user just wants conversation, handle it directly without starting MCP service
if "CONVERSATION" in intent:
# Format conversation history for context
history_context = ""
if history:
history_context = "\nPrevious conversation:\n"
for i, (user_msg, assistant_msg) in enumerate(history[-5:]):
history_context += f"User: {user_msg}\nAssistant: {assistant_msg}\n"
# Create a conversational prompt
conversation_prompt = f"""
You are an intelligent financial analysis assistant with expertise in investment research and financial analysis.
You can engage in natural conversation and provide insights based on your knowledge and the context provided.
{history_context}
Current user message: {message}
Guidelines for your response:
1. If the user is just greeting you or having casual conversation, respond warmly and naturally
2. If the user is asking about a specific financial report or company analysis, explain that you can help search for and analyze financial reports
3. If the user is asking follow-up questions about investments or financial concepts, provide informed insights based on your expertise
4. If the user wants to discuss general financial topics, engage in a knowledgeable discussion
5. Always be helpful, conversational, and friendly while maintaining your expertise
6. Keep responses focused and under 500 words
7. For casual greetings or simple questions, keep your response brief and natural
Please provide a helpful, conversational response:
"""
messages = [
{"role": "system", "content": "You are an intelligent financial analysis assistant with expertise in investment research and financial analysis. You can engage in natural conversation and provide insights based on your knowledge and the context provided. Always be helpful and conversational while maintaining your expertise."},
{"role": "user", "content": conversation_prompt}
]
# Get response from LLM with streaming for better UX
response = client.chat.completions.create(
model="Qwen/Qwen2.5-72B-Instruct",
messages=messages,
max_tokens=min(max_tokens, 2048),
temperature=temperature,
top_p=top_p,
stream=True,
)
# Handle streaming response
conversation_result = ""
for chunk in response:
if hasattr(chunk, 'choices') and len(chunk.choices) > 0:
if hasattr(chunk.choices[0], 'delta') and hasattr(chunk.choices[0].delta, 'content'):
content = chunk.choices[0].delta.content
if content:
conversation_result += content
# Yield partial results for streaming output
output_messages = [conversation_result]
yield "\n".join(output_messages)
return current_session_url, agent_context # Return session URL and context for next turn
except Exception as e:
print(f"[DEBUG] Error in intent check: {str(e)}")
# Continue with normal flow if intent check fails
# If we reach here, user needs tools - start MCP service
output_messages.append("🔄 Starting financial report processing service...")
# ✅ 显示当前使用的EasyReportDataMCP服务配置
print(f"[CONFIG] EasyReportDataMCP: Using Local service at {THIRD_PARTY_MCP_URL}")
yield "\n".join(output_messages)
success, mcp_process = start_mcp_service(hf_token)
if not success:
output_messages.append("❌ Failed to start the financial report processing service. Please check the logs.")
yield "\n".join(output_messages)
return current_session_url, agent_context # Return session URL and context even on failure
try:
# Get available MCP tools
output_messages.append("🔍 Discovering available financial analysis tools...")
yield "\n".join(output_messages)
tools_info = get_available_mcp_tools(mcp_process)
# Let LLM decide which tools to use
output_messages.append("🤖 Analyzing your request and deciding which tools to use...")
yield "\n".join(output_messages)
tool_plan = decide_tool_execution_plan(message, tools_info, history, hf_token, agent_context)
# CRITICAL: Check if plan generation failed due to API error
explanation = tool_plan.get("explanation", "No explanation provided")
plan_list = tool_plan.get("plan", [])
# Debug logging
print(f"[DEBUG] Tool plan explanation: {explanation}")
print(f"[DEBUG] Tool plan list: {plan_list}")
# Check for API errors
if "Error generating plan:" in explanation or "API temporarily unavailable" in explanation:
# Plan generation failed - show error and stop
output_messages.append(f"❌ Unable to process your request: {explanation}")
output_messages.append("")
output_messages.append("💡 Please try again in a moment. This is likely a temporary API issue.")
yield "\n".join(output_messages)
return current_session_url, agent_context # Stop execution to prevent hallucination
# Check if plan is empty
if not plan_list:
print(f"[DEBUG] Empty tool plan received for message: {message}")
# CRITICAL: If explanation is empty AND plan is empty, this is likely an LLM failure
# Check if the original message looks like it was asking for tool usage
# by looking at whether the message would have triggered tool discovery
if explanation == "No explanation provided" or len(explanation.strip()) < 10:
# LLM failed to provide any meaningful response - this is a technical error
output_messages.append("❌ Oops! I encountered a technical issue while processing your request.")
output_messages.append("")
output_messages.append("💡 This could be due to:")
output_messages.append(" • Temporary API service issues")
output_messages.append(" • High system load")
output_messages.append("")
output_messages.append("🔄 Please try again in a moment. If the issue persists, feel free to reach out for support.")
yield "\n".join(output_messages)
return current_session_url, agent_context # CRITICAL: Stop here, don't enter conversation mode
output_messages.append(f'💡 Agent Plan: {explanation}
')
yield "\n".join(output_messages)
# Execute the tool plan
if tool_plan.get("plan"):
tool_results = []
successful_tools = 0
search_returned_no_results = False # 添加标志位
for result in execute_tool_plan(mcp_process, tool_plan, output_messages, message, hf_token):
if isinstance(result, list):
tool_results = result
# Extract successful_tools count from results
for item in tool_results:
if isinstance(item, dict) and "successful_tools" in item:
successful_tools = item["successful_tools"]
# Check if search returned no results
if "search_returned_no_results" in item:
search_returned_no_results = item["search_returned_no_results"]
break
yield "\n".join(output_messages)
# If search returned no results flag is set, directly engage in natural conversation without executing subsequent analysis
if search_returned_no_results:
output_messages.append("💡 No relevant results found from the search, I will engage in natural conversation with you based on existing knowledge.")
yield "\n".join(output_messages)
return current_session_url, agent_context # Return directly without executing subsequent analysis steps
# Check if we have any successful tool results
if successful_tools == 0:
output_messages.append("⚠️ No tools were successfully executed. Unable to provide analysis based on tool results.")
output_messages.append("💡 Please provide a valid financial report URL (PDF format) for analysis.")
yield "\n".join(output_messages)
return current_session_url, agent_context
else:
# No tool plan was generated - check if we should use context for analysis
# CRITICAL: If user is asking for analysis and we have financial data in context, use it!
if agent_context and 'last_financial_data' in agent_context:
# User asked to analyze but we have the data in context already
print(f"[CONTEXT] Using stored financial data for analysis request")
try:
client = InferenceClient(
model="Qwen/Qwen2.5-72B-Instruct",
token=hf_token if hf_token else None
)
# Extract financial data from context
data = agent_context['last_financial_data']
company_name = agent_context.get('last_company_name', 'the company')
period = agent_context.get('last_period', 'the period')
source_url = agent_context.get('last_financial_report_url', '')
# Format the financial data for analysis
financial_summary = f"""Company: {company_name}
Period: {period}
"""
if 'total_revenue' in data:
financial_summary += f"Total Revenue: ${data['total_revenue']:,}\n"
if 'net_income' in data:
financial_summary += f"Net Income: ${data['net_income']:,}\n"
if 'earnings_per_share' in data:
financial_summary += f"Earnings Per Share: ${data['earnings_per_share']}\n"
if 'operating_expenses' in data:
financial_summary += f"Operating Expenses: ${data['operating_expenses']:,}\n"
if 'operating_cash_flow' in data:
financial_summary += f"Operating Cash Flow: ${data['operating_cash_flow']:,}\n"
if source_url:
financial_summary += f"\nSource: {source_url}\n"
# Create analysis prompt
analysis_prompt = f"""
You are a professional financial analyst. Analyze the following financial report data and provide comprehensive investment insights.
{financial_summary}
Additional data details:
{json.dumps(data, indent=2)}
User's analysis request: {message}
Please provide a detailed analysis covering:
1. Revenue performance and trends
2. Profitability analysis (net income, margins, ROE, etc.)
3. Operating efficiency (expense ratios, cash flow)
4. Key financial metrics interpretation
5. Investment recommendations and risk assessment
6. Specific insights based on the user's request
Provide specific numbers and percentages from the data. Be detailed and data-driven.
IMPORTANT: Use ONLY the actual numbers provided above - DO NOT make up or hallucinate any financial figures.
"""
output_messages.append("📊 Analyzing financial data from context...")
yield "\n".join(output_messages)
messages = [
{"role": "system", "content": "You are a professional financial analyst providing detailed investment insights based on financial reports. Always use actual data from the reports and provide specific numbers. Never hallucinate or make up financial figures."},
{"role": "user", "content": analysis_prompt}
]
# Get response from LLM with streaming
response = client.chat.completions.create(
model="Qwen/Qwen2.5-72B-Instruct",
messages=messages,
max_tokens=min(max_tokens, 2048),
temperature=0.3, # Lower temperature for more factual analysis
top_p=top_p,
stream=True,
)
# Handle streaming response
analysis_result = ""
output_messages.append("") # Empty line before analysis
for chunk in response:
if hasattr(chunk, 'choices') and len(chunk.choices) > 0:
if hasattr(chunk.choices[0], 'delta') and hasattr(chunk.choices[0].delta, 'content'):
content = chunk.choices[0].delta.content
if content:
analysis_result += content
output_messages[-1] = analysis_result
yield "\n".join(output_messages)
# Return after analysis
return current_session_url, agent_context
except Exception as e:
error_msg = f"❌ Error during analysis: {str(e)}"
print(f"[DEBUG] {error_msg}")
output_messages.append(error_msg)
yield "\n".join(output_messages)
return current_session_url, agent_context
# If no context data, engage in natural conversation
try:
client = InferenceClient(
model="Qwen/Qwen2.5-72B-Instruct",
token=hf_token if hf_token else None
)
# Format conversation history for context
history_context = ""
if history:
history_context = "\nPrevious conversation:\n"
for i, (user_msg, assistant_msg) in enumerate(history[-5:]): # Include last 5 exchanges
history_context += f"User: {user_msg}\nAssistant: {assistant_msg}\n"
# Create a conversational prompt
conversation_prompt = f"""
You are an intelligent financial analysis assistant with expertise in investment research and financial analysis.
You can engage in natural conversation and provide insights based on your knowledge and the context provided.
{history_context}
Current user message: {message}
Guidelines for your response:
1. If the user is just greeting you or having casual conversation, respond warmly and naturally
2. If the user is asking about a specific financial report or company analysis, explain that you can help search for and analyze financial reports
3. If the user is asking follow-up questions about investments or financial concepts, provide informed insights based on your expertise
4. If the user wants to discuss general financial topics, engage in a knowledgeable discussion
5. Always be helpful, conversational, and friendly while maintaining your expertise
6. Keep responses focused and under 500 words
7. For casual greetings or simple questions, keep your response brief and natural
Please provide a helpful, conversational response:
"""
messages = [
{"role": "system", "content": "You are an intelligent financial analysis assistant with expertise in investment research and financial analysis. You can engage in natural conversation and provide insights based on your knowledge and the context provided. Always be helpful and conversational while maintaining your expertise."},
{"role": "user", "content": conversation_prompt}
]
# Get response from LLM with streaming for better UX
response = client.chat.completions.create(
model="Qwen/Qwen2.5-72B-Instruct",
messages=messages,
max_tokens=min(max_tokens, 2048),
temperature=temperature,
top_p=top_p,
stream=True,
)
# Handle streaming response
conversation_result = ""
for chunk in response:
if hasattr(chunk, 'choices') and len(chunk.choices) > 0:
if hasattr(chunk.choices[0], 'delta') and hasattr(chunk.choices[0].delta, 'content'):
content = chunk.choices[0].delta.content
if content:
conversation_result += content
# Yield partial results for streaming output
if not output_messages or output_messages[-1] != conversation_result:
if output_messages and output_messages[-1].startswith("💡 No specific tools needed"):
output_messages[-1] = conversation_result
else:
output_messages.append(conversation_result)
else:
output_messages[-1] = conversation_result
yield "\n".join(output_messages)
except Exception as e:
error_msg = f"❌ Error during conversation: {str(e)}"
print(f"[DEBUG] {error_msg}")
output_messages.append(error_msg)
yield "\n".join(output_messages)
# Return after conversation - no need to process tool results
return current_session_url, agent_context
# Filter out the successful_tools item from tool_results
filtered_tool_results = [result for result in tool_results if not (isinstance(result, dict) and "successful_tools" in result)] if 'tool_results' in locals() else []
# Debug: Print all tool results
# print(f"[DEBUG] All tool results: {json.dumps(filtered_tool_results, indent=2)}")
# Intelligent result processing - choose the best way to present results to user
# Check for file analysis trigger or URL analysis trigger
analysis_file_path = None
analysis_url = None
# Look for file analysis triggers in tool results
for result in filtered_tool_results:
if result is not None and 'tool' in result and result['tool'] == 'analyze_financial_report_file':
# Check for structuredContent format (newer MCP responses)
tool_result_data = None
if 'result' in result and result['result'] is not None:
# NEW: Check for MCP 'content' array format (most common)
if 'content' in result['result'] and isinstance(result['result']['content'], list) and len(result['result']['content']) > 0:
# Extract text from content array
content_item = result['result']['content'][0]
if isinstance(content_item, dict) and 'text' in content_item:
try:
# Parse JSON string
tool_result_data = json.loads(content_item['text'])
print(f"[DEBUG] Parsed analyze_financial_report_file result from content array")
except json.JSONDecodeError as e:
print(f"[DEBUG] Failed to parse JSON from analyze_financial_report_file content: {str(e)}")
tool_result_data = None
# Fallback: Check for structuredContent format (older MCP responses)
elif 'structuredContent' in result['result'] and 'result' in result['result']['structuredContent']:
tool_result_data = result['result']['structuredContent']['result']
# Fallback: Check for direct result format (oldest MCP responses)
elif 'type' in result['result']:
tool_result_data = result['result']
if tool_result_data and isinstance(tool_result_data, dict):
if tool_result_data.get('type') == 'file_analysis_trigger':
print(f"[DEBUG] Found file_analysis_trigger, keys: {list(tool_result_data.keys())}")
print(f"[DEBUG] Has 'content': {'content' in tool_result_data}")
print(f"[DEBUG] Has 'filename': {'filename' in tool_result_data}")
if 'content' in tool_result_data and 'filename' in tool_result_data:
# We have file content - perform streaming analysis
file_content = tool_result_data['content']
filename = tool_result_data['filename']
source_url = tool_result_data.get('source_url', '') # Get source URL if available
# CRITICAL: If no source URL in current result, use session URL
if not source_url and current_session_url:
source_url = current_session_url
print(f"[SESSION] Using session URL for analysis: {source_url}")
print(f"[ANALYSIS] Analyzing content: {len(file_content)} characters")
output_messages.append("\n📊 Generating financial analysis...")
yield "\n".join(output_messages)
# Build analysis prompt
source_context = ""
if source_url:
source_context = f"\n\nSource: {source_url}"
analysis_prompt = f"""
You are a professional financial analyst. Analyze the following financial report and provide comprehensive investment insights.
Financial Report: {filename}{source_context}
Report Content:
{file_content}
Please provide a detailed analysis covering:
1. Revenue performance and trends
2. Profitability analysis (net income, margins, etc.)
3. Balance sheet health (assets, liabilities, debt ratios)
4. Cash flow analysis
5. Key financial ratios and metrics
6. Investment recommendations and risk assessment
Provide specific numbers and percentages from the report. Be detailed and data-driven.
"""
# CRITICAL: Create InferenceClient OUTSIDE the HTML detection block
try:
client = InferenceClient(
model="Qwen/Qwen2.5-72B-Instruct",
token=hf_token if hf_token else None
)
messages = [
{"role": "system", "content": "You are a professional financial analyst providing detailed investment insights based on financial reports. Always use actual data from the reports and provide specific numbers."},
{"role": "user", "content": analysis_prompt}
]
# Stream the analysis results
output_messages.append("") # Add empty line
yield "\n".join(output_messages)
response = client.chat.completions.create(
model="Qwen/Qwen2.5-72B-Instruct",
messages=messages,
max_tokens=4000, # Increased from 2000 to 4000 for complete analysis
temperature=0.3,
stream=True, # Enable streaming!
)
# Handle streaming response
analysis_result = ""
for chunk in response:
if hasattr(chunk, 'choices') and len(chunk.choices) > 0:
if hasattr(chunk.choices[0], 'delta') and hasattr(chunk.choices[0].delta, 'content'):
content = chunk.choices[0].delta.content
if content:
analysis_result += content
# Update with streaming content
if output_messages and output_messages[-1] == "":
output_messages.append(analysis_result)
else:
output_messages[-1] = analysis_result
yield "\n".join(output_messages)
# Analysis complete
return current_session_url, agent_context
except Exception as e:
print(f"[ERROR] Failed to analyze file: {str(e)}")
output_messages.append(f"\n⚠️ Analysis failed: {str(e)}")
yield "\n".join(output_messages)
return current_session_url, agent_context
# Old format - just store file path
if 'file_path' in tool_result_data:
# Fix the file path - MCP service stores files in MCP_Financial_Report/financial_reports/
file_path = tool_result_data['file_path']
if file_path.startswith('financial_reports/'):
corrected_file_path = f"MCP_Financial_Report/{file_path}"
analysis_file_path = corrected_file_path
print(f"[DEBUG] Found file analysis trigger with corrected file_path: {analysis_file_path}")
else:
analysis_file_path = file_path
print(f"[DEBUG] Found file analysis trigger with file_path: {analysis_file_path}")
break
elif tool_result_data.get('type') == 'url_analysis_trigger':
if 'url' in tool_result_data:
analysis_url = tool_result_data['url']
print(f"[DEBUG] Found URL analysis trigger with url: {analysis_url}")
break
# Also check the old format for backward compatibility
if not analysis_file_path and not analysis_url:
for result in filtered_tool_results:
if result is not None and 'tool' in result and result['tool'] == 'analyze_financial_report_file':
if 'file_path' in result['result']:
# Fix the file path - MCP service stores files in MCP_Financial_Report/financial_reports/
file_path = result['result']['file_path']
if file_path.startswith('financial_reports/'):
corrected_file_path = f"MCP_Financial_Report/{file_path}"
analysis_file_path = corrected_file_path
print(f"[DEBUG] Found old format file analysis with corrected file_path: {analysis_file_path}")
else:
analysis_file_path = file_path
print(f"[DEBUG] Found old format file analysis with file_path: {analysis_file_path}")
break
# If we still don't have a file path, check if we downloaded a file and use that
if not analysis_file_path and not analysis_url:
for result in filtered_tool_results:
if result is not None and 'tool' in result and result['tool'] == 'download_financial_report':
# Check for structuredContent format (newer MCP responses)
if 'result' in result and result['result'] is not None:
download_result = None
if 'structuredContent' in result['result'] and 'result' in result['result']['structuredContent']:
download_result = result['result']['structuredContent']['result']
# Check for direct result format (older MCP responses)
elif isinstance(result['result'], dict) and 'filename' in result['result']:
download_result = result['result']
if download_result and isinstance(download_result, dict) and 'filepath' in download_result:
# Fix the file path - MCP service stores files in MCP_Financial_Report/financial_reports/
file_path = download_result['filepath']
if file_path.startswith('financial_reports/'):
corrected_file_path = f"MCP_Financial_Report/{file_path}"
analysis_file_path = corrected_file_path
print(f"[DEBUG] Found download result with corrected file_path: {analysis_file_path}")
else:
analysis_file_path = file_path
print(f"[DEBUG] Found download result with file_path: {analysis_file_path}")
break
# Debug: Print filtered_tool_results summary (simplified)
print(f"[DEBUG] ===== FILTERED TOOL RESULTS =====")
print(f"[DEBUG] Total filtered_tool_results: {len(filtered_tool_results)}")
for i, result in enumerate(filtered_tool_results):
tool_name = result.get('tool', 'unknown') if result else 'None'
print(f"[DEBUG] Result {i}: tool={tool_name}")
# Check for final_download_link results first
has_download_links = False
download_links_content = []
print(f"[DEBUG] Checking {len(filtered_tool_results)} filtered tool results for download links")
for result in filtered_tool_results:
# Simplified debug output - only show tool name and type
if result is not None and 'tool' in result:
tool_name = result.get('tool', 'unknown')
print(f"[DEBUG] Processing tool: {tool_name}")
if result is not None and 'tool' in result and 'result' in result and result['result'] is not None:
# Check for structuredContent format (newer MCP responses)
tool_result_data = None
# NEW: Check for MCP 'content' array format (most common)
if 'content' in result['result'] and isinstance(result['result']['content'], list) and len(result['result']['content']) > 0:
# Extract text from content array
content_item = result['result']['content'][0]
if isinstance(content_item, dict) and 'text' in content_item:
try:
# Parse JSON string
tool_result_data = json.loads(content_item['text'])
print(f"[DEBUG] Parsed tool result from content array")
except json.JSONDecodeError as e:
print(f"[DEBUG] Failed to parse JSON from content: {str(e)}")
tool_result_data = None
# Fallback: Check for structuredContent format
elif 'structuredContent' in result['result'] and 'result' in result['result']['structuredContent']:
tool_result_data = result['result']['structuredContent']['result']
print(f"[DEBUG] Extracted tool result from structuredContent")
# Fallback: Check for direct result format (older MCP responses)
elif isinstance(result['result'], dict):
tool_result_data = result['result']
print(f"[DEBUG] Using result dict directly")
print(f"[DEBUG] Tool: {result.get('tool', 'unknown')}, Type: {tool_result_data.get('type') if tool_result_data else 'None'}")
if tool_result_data and isinstance(tool_result_data, dict):
# Handle final_download_link type
if tool_result_data.get('type') == 'final_download_link':
has_download_links = True
download_links_content.append({
"title": tool_result_data.get('title', 'Financial Report'),
"link": tool_result_data.get('link', ''),
"snippet": tool_result_data.get('snippet', '')
})
print(f"[DEBUG] Found final_download_link: {tool_result_data.get('title')}")
# Handle download_link_extracted type (single link) - THIS WAS MISSING!
elif tool_result_data.get('type') == 'download_link_extracted':
has_download_links = True
download_links_content.append({
"title": tool_result_data.get('title', 'Financial Report'),
"link": tool_result_data.get('link', ''),
"snippet": tool_result_data.get('snippet', '')
})
print(f"[DEBUG] Found download_link_extracted: {tool_result_data.get('title')}")
# Handle download_links_extracted type (multiple links)
elif tool_result_data.get('type') == 'download_links_extracted' and 'links' in tool_result_data:
has_download_links = True
for link_info in tool_result_data['links']:
download_links_content.append({
"title": link_info.get('title', 'Financial Report'),
"link": link_info.get('url', link_info.get('link', '')),
"snippet": link_info.get('snippet', '')
})
print(f"[DEBUG] Found download_links_extracted: {len(tool_result_data['links'])} links")
print(f"[DEBUG] Total download links found: {len(download_links_content)}")
# If we have download links, generate intelligent final response
if has_download_links:
print(f"[DEBUG] ===== ENTERING DOWNLOAD LINKS BRANCH =====")
print(f"[DEBUG] Found {len(download_links_content)} download link(s)")
# print(f"[DEBUG] download_links_content: {json.dumps(download_links_content, indent=2)}")
# First, show the download links in the execution log
output_messages.append("\n✅ Download link(s) found successfully!")
yield "\n".join(output_messages)
# Now generate an intelligent final Agent response
try:
client = InferenceClient(
model="Qwen/Qwen2.5-72B-Instruct",
token=hf_token if hf_token else None
)
# Format the download links for the prompt
links_summary = ""
for i, link_info in enumerate(download_links_content, 1):
links_summary += f"\n{i}. Title: {link_info['title']}\n URL: {link_info['link']}\n"
if link_info['snippet']:
links_summary += f" Description: {link_info['snippet']}\n"
# Create prompt for final response
final_response_prompt = f"""
You are a helpful financial analysis assistant. Based on the user's request and the tool execution results, provide a clear, concise, and intelligent final response.
User's original request: {message}
Tool execution results - Download links found:
{links_summary}
IMPORTANT INSTRUCTIONS:
1. Analyze the user's request carefully to understand their true intent
2. You MUST use the EXACT URLs provided in the tool results above - DO NOT modify or invent URLs
3. Decide intelligently how to present the information based on what the user asked for:
- If they want ONE link, select the most relevant one
- If they want to compare/analyze multiple reports, present relevant options
- If they want specific information, provide that with supporting links
4. Present information clearly with proper markdown formatting
5. Use emoji appropriately (📄 for title, 🔗 for URL, 📋 for description)
6. Keep your response helpful and aligned with the user's actual intent
7. DO NOT make assumptions - let the user's question guide your response format
Provide an intelligent, contextual response:
"""
messages = [
{"role": "system", "content": "You are a helpful financial analysis assistant that provides clear and concise responses based on tool execution results."},
{"role": "user", "content": final_response_prompt}
]
# Add separator for final response
output_messages.append("")
yield "\n".join(output_messages)
# Get streaming response from LLM
response = client.chat.completions.create(
model="Qwen/Qwen2.5-72B-Instruct",
messages=messages,
max_tokens=min(max_tokens, 1000),
temperature=0.7,
top_p=top_p,
stream=True,
)
# Handle streaming response
final_answer = ""
for chunk in response:
if hasattr(chunk, 'choices') and len(chunk.choices) > 0:
if hasattr(chunk.choices[0], 'delta') and hasattr(chunk.choices[0].delta, 'content'):
content = chunk.choices[0].delta.content
if content:
final_answer += content
# Update the last message with streaming content
if output_messages and output_messages[-1] == "":
output_messages.append(final_answer)
else:
output_messages[-1] = final_answer
yield "\n".join(output_messages)
except Exception as e:
print(f"[DEBUG] Error generating final response: {str(e)}")
# Fallback to simple presentation if LLM fails
output_messages.append("\n💡 I've found the download link(s) you requested:")
output_messages.append("")
for link_info in download_links_content:
output_messages.append(f"📄 **{link_info['title']}**")
output_messages.append(f"🔗 {link_info['link']}")
if link_info['snippet']:
output_messages.append(f"📋 {link_info['snippet']}")
output_messages.append("")
output_messages.append("✅ You can click on the links above to download the financial reports directly.")
yield "\n".join(output_messages)
return current_session_url, agent_context # End processing here - user just wanted download links
# Fallback: Present tool results summary and generate intelligent final response
elif filtered_tool_results:
# Re-check if there are any download links in the results that we might have missed
additional_download_links = []
for result in filtered_tool_results:
if result is not None and 'tool' in result and 'result' in result and result['result'] is not None:
tool_result_data = None
# NEW: Check for MCP 'content' array format (most common)
if 'content' in result['result'] and isinstance(result['result']['content'], list) and len(result['result']['content']) > 0:
# Extract text from content array
content_item = result['result']['content'][0]
if isinstance(content_item, dict) and 'text' in content_item:
try:
# Parse JSON string
tool_result_data = json.loads(content_item['text'])
except json.JSONDecodeError:
tool_result_data = None
# Fallback: Check for structuredContent format
elif 'structuredContent' in result['result'] and 'result' in result['result']['structuredContent']:
tool_result_data = result['result']['structuredContent']['result']
# Fallback: Check for direct result format
elif isinstance(result['result'], dict):
tool_result_data = result['result']
# print(f"[DEBUG] Checking tool result data: {json.dumps(tool_result_data, indent=2) if tool_result_data else 'None'}")
if tool_result_data and isinstance(tool_result_data, dict):
# Check for download_link_extracted type (THIS WAS MISSING!)
if tool_result_data.get('type') == 'download_link_extracted':
additional_download_links.append({
"title": tool_result_data.get('title', 'Financial Report'),
"link": tool_result_data.get('link', ''),
"snippet": tool_result_data.get('snippet', '')
})
# Check if this result contains links array
elif 'links' in tool_result_data and isinstance(tool_result_data['links'], list):
for link_info in tool_result_data['links']:
if isinstance(link_info, dict) and 'url' in link_info:
additional_download_links.append({
"title": link_info.get('title', 'Financial Report'),
"link": link_info.get('url', ''),
"snippet": link_info.get('snippet', '')
})
# Check if this result contains a single link/url field
elif 'link' in tool_result_data or 'url' in tool_result_data:
additional_download_links.append({
"title": tool_result_data.get('title', 'Financial Report'),
"link": tool_result_data.get('link', tool_result_data.get('url', '')),
"snippet": tool_result_data.get('snippet', '')
})
print(f"[DEBUG] Found {len(additional_download_links)} additional download links")
# If we found download links in this branch, process them
if additional_download_links:
output_messages.append("\n✅ Download link(s) found successfully!")
yield "\n".join(output_messages)
try:
client = InferenceClient(
model="Qwen/Qwen2.5-72B-Instruct",
token=hf_token if hf_token else None
)
# Format the download links for the prompt
links_summary = ""
for i, link_info in enumerate(additional_download_links, 1):
links_summary += f"\n{i}. Title: {link_info['title']}\n URL: {link_info['link']}\n"
if link_info['snippet']:
links_summary += f" Description: {link_info['snippet']}\n"
# Create prompt for final response
final_response_prompt = f"""
You are a helpful financial analysis assistant. Based on the user's request and the tool execution results, provide a clear, concise, and intelligent final response.
User's original request: {message}
Tool execution results - Download links found:
{links_summary}
IMPORTANT INSTRUCTIONS:
1. Analyze the user's request carefully to understand their true intent
2. You MUST use the EXACT URLs provided in the tool results above - DO NOT modify or invent URLs
3. Decide intelligently how to present the information based on what the user asked for:
- If they want ONE link, select the most relevant one
- If they want to compare/analyze multiple reports, present relevant options
- If they requested a TABLE format, use markdown table syntax
- If they want specific information, provide that with supporting links
4. Present information clearly with proper markdown formatting
5. Use emoji appropriately (📄 for title, 🔗 for URL, 📋 for description)
6. Keep your response helpful and aligned with the user's actual intent
7. DO NOT make assumptions - let the user's question guide your response format
8. For table format, use markdown table like:
| Column 1 | Column 2 |
|----------|----------|
| Data 1 | Data 2 |
Provide an intelligent, contextual response:
"""
messages = [
{"role": "system", "content": "You are a helpful financial analysis assistant that provides clear and concise responses based on tool execution results."},
{"role": "user", "content": final_response_prompt}
]
# Add separator for final response
output_messages.append("")
yield "\n".join(output_messages)
# Get streaming response from LLM
response = client.chat.completions.create(
model="Qwen/Qwen2.5-72B-Instruct",
messages=messages,
max_tokens=min(max_tokens, 1000),
temperature=0.7,
top_p=top_p,
stream=True,
)
# Handle streaming response
final_answer = ""
for chunk in response:
if hasattr(chunk, 'choices') and len(chunk.choices) > 0:
if hasattr(chunk.choices[0], 'delta') and hasattr(chunk.choices[0].delta, 'content'):
content = chunk.choices[0].delta.content
if content:
final_answer += content
# Update the last message with streaming content
if output_messages and output_messages[-1] == "":
output_messages.append(final_answer)
else:
output_messages[-1] = final_answer
yield "\n".join(output_messages)
return # End processing here
except Exception as e:
print(f"[DEBUG] Error generating final response: {str(e)}")
# Fallback to simple presentation if LLM fails
output_messages.append("\n💡 I've found the download link(s) you requested:")
output_messages.append("")
for link_info in additional_download_links[:1]: # Show only the first one
output_messages.append(f"📄 **{link_info['title']}**")
output_messages.append(f"🔗 {link_info['link']}")
if link_info['snippet']:
output_messages.append(f"📋 {link_info['snippet']}")
output_messages.append("")
yield "\n".join(output_messages)
return
# If no download links found, collect tool execution summary
# First, collect tool execution summary including actual data
tool_summary = ""
has_error_results = False # Flag to track if any tools returned errors
successful_data_retrieval = False # Flag to track if we successfully retrieved data
for result in filtered_tool_results:
# Check if this tool execution was marked as failed
if isinstance(result, dict) and result.get("success") == False:
tool_name = result.get("tool", "")
# Check if this is a third-party MCP tool
if tool_name in THIRD_PARTY_MCP_TOOLS:
# Only consider it an error if it's a critical tool or if we haven't successfully retrieved data yet
# Critical tools are those that directly retrieve financial data
critical_tools = ["get_financial_data", "extract_financial_metrics", "get_latest_financial_data"]
if tool_name in critical_tools or not successful_data_retrieval:
has_error_results = True
error_info = result.get("result", {})
error_msg = error_info.get("error", "Unknown error") if isinstance(error_info, dict) else str(error_info)
print(f"[DEBUG] Third-party tool {tool_name} failed: {error_msg}")
# Check if this tool execution was successful and retrieved financial data
elif isinstance(result, dict) and result.get("success") == True:
tool_name = result.get("tool", "")
# Check if this tool retrieved financial data
if tool_name in ["get_financial_data", "extract_financial_metrics", "get_latest_financial_data"]:
tool_result = result.get("result", {})
# Check if the result contains actual financial data
if tool_result and isinstance(tool_result, dict):
# Look for financial data indicators in various result formats
has_financial_data = False
# Check direct result format
if "period" in tool_result and ("total_revenue" in tool_result or "net_income" in tool_result):
has_financial_data = True
# Check content array format
elif "content" in tool_result and isinstance(tool_result["content"], list) and len(tool_result["content"]) > 0:
content_item = tool_result["content"][0]
if isinstance(content_item, dict) and "text" in content_item:
try:
content_json = json.loads(content_item["text"])
if isinstance(content_json, dict) and ("period" in content_json and ("total_revenue" in content_json or "net_income" in content_json)):
has_financial_data = True
except json.JSONDecodeError:
pass # Not JSON, continue normally
if has_financial_data:
successful_data_retrieval = True
has_error_results = False # Override any previous errors since we got the data
print(f"[DEBUG] Successfully retrieved financial data from {tool_name}")
# Also check for errors in successful tool results (tools that ran but returned error data)
if result is not None and 'tool' in result and 'result' in result and result['result'] is not None:
tool_name = result.get('tool', '')
tool_result = result.get('result', {})
# Check if this is a third-party MCP tool
if tool_name in THIRD_PARTY_MCP_TOOLS:
# Check for error in the result
if isinstance(tool_result, dict):
# Check direct error field
if 'error' in tool_result and tool_result['error']:
# Only consider it an error if we haven't successfully retrieved data yet
if not successful_data_retrieval:
has_error_results = True
print(f"[DEBUG] Third-party tool {tool_name} returned error: {tool_result['error']}")
# Check structured content for errors
elif 'structuredContent' in tool_result and 'result' in tool_result['structuredContent']:
structured_result = tool_result['structuredContent']['result']
if isinstance(structured_result, dict) and 'error' in structured_result and structured_result['error']:
# Only consider it an error if we haven't successfully retrieved data yet
if not successful_data_retrieval:
has_error_results = True
print(f"[DEBUG] Third-party tool {tool_name} returned error in structuredContent: {structured_result['error']}")
# Check content array for errors
elif 'content' in tool_result and isinstance(tool_result['content'], list) and len(tool_result['content']) > 0:
content_item = tool_result['content'][0]
if isinstance(content_item, dict) and 'text' in content_item:
try:
content_json = json.loads(content_item['text'])
if isinstance(content_json, dict) and 'error' in content_json and content_json['error']:
# Only consider it an error if we haven't successfully retrieved data yet
if not successful_data_retrieval:
has_error_results = True
print(f"[DEBUG] Third-party tool {tool_name} returned error in content: {content_json['error']}")
except json.JSONDecodeError:
pass # Not JSON, continue normally
# If any third-party tools returned errors, don't generate fake data
if has_error_results:
output_messages.append("\n❌ Some tools encountered errors. Unable to provide accurate financial data.")
output_messages.append("💡 This may be because the requested data doesn't exist or there was an issue accessing the SEC database.")
yield "\n".join(output_messages)
return current_session_url, agent_context
for result in filtered_tool_results:
if result is not None and 'tool' in result:
tool_name = result.get('tool', 'Unknown Tool')
tool_summary += f"\nTool: {tool_name}\n"
# Extract result data
if 'result' in result and result['result'] is not None:
tool_result_data = None
# NEW: Check for MCP 'content' array format (most common)
if 'content' in result['result'] and isinstance(result['result']['content'], list) and len(result['result']['content']) > 0:
# Extract text from content array
content_item = result['result']['content'][0]
if isinstance(content_item, dict) and 'text' in content_item:
try:
# Parse JSON string
tool_result_data = json.loads(content_item['text'])
except json.JSONDecodeError:
tool_result_data = None
# Fallback: Check for structuredContent format
elif 'structuredContent' in result['result'] and 'result' in result['result']['structuredContent']:
tool_result_data = result['result']['structuredContent']['result']
# Fallback: Check for direct result format
elif isinstance(result['result'], dict):
tool_result_data = result['result']
if tool_result_data:
# CRITICAL: Check if this result contains error information
# Many third-party tools return errors in the content as JSON strings
if isinstance(tool_result_data, dict) and 'error' in tool_result_data and tool_result_data['error']:
has_error_results = True
tool_summary += f"Error: {tool_result_data['error']}\n"
elif isinstance(tool_result_data, dict) and 'content' in tool_result_data:
# Check if content contains error information
content_items = tool_result_data['content']
if isinstance(content_items, list) and len(content_items) > 0:
first_item = content_items[0]
if isinstance(first_item, dict) and 'text' in first_item:
try:
content_json = json.loads(first_item['text'])
if isinstance(content_json, dict) and 'error' in content_json and content_json['error']:
has_error_results = True
tool_summary += f"Error: {content_json['error']}\n"
except json.JSONDecodeError:
pass # Not JSON, continue normally
# CRITICAL: Include source_url if available (especially for financial data)
if 'source_url' in tool_result_data and tool_result_data['source_url']:
tool_summary += f"Source URL: {tool_result_data['source_url']}\n"
# Include other financial data fields
if 'period' in tool_result_data:
tool_summary += f"Period: {tool_result_data['period']}\n"
if 'total_revenue' in tool_result_data:
tool_summary += f"Revenue: ${tool_result_data['total_revenue']:,.0f}\n"
if 'net_income' in tool_result_data:
tool_summary += f"Net Income: ${tool_result_data['net_income']:,.0f}\n"
if 'earnings_per_share' in tool_result_data:
tool_summary += f"EPS: ${tool_result_data['earnings_per_share']}\n"
# Show summary information
if 'message' in tool_result_data:
tool_summary += f"Message: {tool_result_data['message']}\n"
if 'type' in tool_result_data:
tool_summary += f"Type: {tool_result_data['type']}\n"
# CRITICAL: Include actual links data so LLM can see them!
if 'links' in tool_result_data and isinstance(tool_result_data['links'], list):
tool_summary += "Links found:\n"
for i, link_info in enumerate(tool_result_data['links'], 1):
if isinstance(link_info, dict):
tool_summary += f" {i}. Title: {link_info.get('title', 'N/A')}\n"
tool_summary += f" URL: {link_info.get('url', link_info.get('link', 'N/A'))}\n"
if 'snippet' in link_info:
tool_summary += f" Description: {link_info['snippet']}\n"
elif 'link' in tool_result_data or 'url' in tool_result_data:
tool_summary += f"Link: {tool_result_data.get('link', tool_result_data.get('url', 'N/A'))}\n"
if 'title' in tool_result_data:
tool_summary += f"Title: {tool_result_data['title']}\n"
if 'snippet' in tool_result_data:
tool_summary += f"Description: {tool_result_data['snippet']}\n"
output_messages.append("\n✅ Tool execution completed successfully!")
yield "\n".join(output_messages)
# CRITICAL: Extract key information from tool results to agent context for multi-turn dialogue
for result in filtered_tool_results:
if result is not None and 'tool' in result and 'result' in result and result['result'] is not None:
tool_name = result.get('tool', '')
tool_result = result.get('result', {})
# Extract text from MCP content format
if 'content' in tool_result and isinstance(tool_result['content'], list) and len(tool_result['content']) > 0:
content_item = tool_result['content'][0]
if isinstance(content_item, dict) and 'text' in content_item:
try:
parsed_data = json.loads(content_item['text'])
# Extract company information from search_company
if tool_name == 'search_company' and isinstance(parsed_data, dict):
if 'cik' in parsed_data:
agent_context['last_company_cik'] = parsed_data['cik']
print(f"[CONTEXT] Stored CIK: {parsed_data['cik']}")
if 'name' in parsed_data:
agent_context['last_company_name'] = parsed_data['name']
print(f"[CONTEXT] Stored company name: {parsed_data['name']}")
if 'ticker' in parsed_data:
agent_context['last_company_ticker'] = parsed_data['ticker']
print(f"[CONTEXT] Stored ticker: {parsed_data['ticker']}")
# Extract financial data from get_financial_data
elif tool_name == 'get_financial_data' and isinstance(parsed_data, dict):
# Check if we have actual financial data or just period
has_financial_data = any(key in parsed_data for key in ['total_revenue', 'net_income', 'earnings_per_share', 'source_url'])
if not has_financial_data:
print(f"[CONTEXT] ⚠️ WARNING: get_financial_data returned incomplete data (only period)")
print(f"[CONTEXT] This likely means the requested financial data is not available")
# Store a flag indicating incomplete data
agent_context['incomplete_financial_data'] = True
agent_context['incomplete_data_reason'] = f"No financial metrics found for {parsed_data.get('period', 'requested period')}"
else:
# We have complete data
agent_context['incomplete_financial_data'] = False
if 'period' in parsed_data:
agent_context['last_period'] = parsed_data['period']
print(f"[CONTEXT] Stored period: {parsed_data['period']}")
if 'total_revenue' in parsed_data:
agent_context['last_revenue'] = parsed_data['total_revenue']
if 'net_income' in parsed_data:
agent_context['last_net_income'] = parsed_data['net_income']
if 'source_url' in parsed_data:
agent_context['last_financial_report_url'] = parsed_data['source_url']
print(f"[CONTEXT] Stored financial report URL: {parsed_data['source_url']}")
# Store the complete financial data for reference
agent_context['last_financial_data'] = parsed_data
print(f"[CONTEXT] Stored financial data for {agent_context.get('last_company_name', 'company')}")
except json.JSONDecodeError:
pass
# Generate intelligent final response based on tool results
try:
# DEBUG: Print the tool_summary to see what LLM receives
print(f"[DEBUG] Tool summary being sent to LLM:")
print(f"=" * 80)
print(tool_summary)
print(f"=" * 80)
# Check if we have incomplete financial data
if agent_context.get('incomplete_financial_data', False):
# Provide a helpful error message to user
company_name = agent_context.get('last_company_name', 'the company')
period = agent_context.get('last_period', 'the requested period')
output_messages.append("")
output_messages.append(f"⚠️ Sorry, detailed financial data for {company_name} {period} is not available in the SEC EDGAR database.")
output_messages.append("")
output_messages.append("💡 This could be because:")
output_messages.append(f" • {company_name}'s fiscal {period} report hasn't been filed yet")
output_messages.append(f" • {company_name} uses a different fiscal calendar")
output_messages.append(f" • The period name might be different (try '2024Q4' or specific fiscal year periods)")
output_messages.append("")
output_messages.append("🔍 You can try:")
output_messages.append(f" • Searching for a different quarter (e.g., '2024Q4', '2024Q3')")
output_messages.append(f" • Visiting the SEC EDGAR website directly: https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany&CIK={agent_context.get('last_company_cik', '')}&type=10-Q&dateb=&owner=exclude&count=40")
yield "\n".join(output_messages)
return current_session_url, agent_context
client = InferenceClient(
model="Qwen/Qwen2.5-72B-Instruct",
token=hf_token if hf_token else None
)
# Create prompt for final response
final_response_prompt = f"""
You are a helpful financial analysis assistant. Based on the user's request and the tool execution results, provide a clear, concise, and intelligent final response.
User's original request: {message}
Tool execution results:
{tool_summary}
IMPORTANT INSTRUCTIONS:
1. Carefully analyze the user's request to understand their true intent
2. The tool execution results above contain the actual data - use them!
3. If a "Source URL" field is provided in the results, YOU MUST include it in your response as a clickable link
4. DO NOT make up or invent any information that is not in the results
5. DO NOT create fake URLs, links, or placeholder links like "Apple 2025 Q1 Financial Report" - only use EXACT URLs from the tool results
6. If financial data is provided with a source_url, format it as: "For more details, see the [official SEC filing](EXACT_URL_HERE)"
7. If the user requested a specific format (e.g., table), provide it using markdown
8. Present information clearly based on what the user actually asked for
9. If results contain links, present them properly formatted with titles and EXACT URLs from the tool results
10. Keep your response helpful and aligned with the user's actual intent
11. CRITICAL: Never create placeholder or fake links - if no URL is in the results, don't include a link
Provide a clear, accurate final response based on the tool execution results above:
"""
messages = [
{"role": "system", "content": "You are a helpful financial analysis assistant that provides clear and concise responses based on tool execution results."},
{"role": "user", "content": final_response_prompt}
]
# Add separator for final response
output_messages.append("")
yield "\n".join(output_messages)
# Get streaming response from LLM
response = client.chat.completions.create(
model="Qwen/Qwen2.5-72B-Instruct",
messages=messages,
max_tokens=min(max_tokens, 1000),
temperature=0.7,
top_p=top_p,
stream=True,
)
# Handle streaming response
final_answer = ""
for chunk in response:
if hasattr(chunk, 'choices') and len(chunk.choices) > 0:
if hasattr(chunk.choices[0], 'delta') and hasattr(chunk.choices[0].delta, 'content'):
content = chunk.choices[0].delta.content
if content:
final_answer += content
# Update the last message with streaming content
if output_messages and output_messages[-1] == "":
output_messages.append(final_answer)
else:
output_messages[-1] = final_answer
yield "\n".join(output_messages)
except Exception as e:
print(f"[DEBUG] Error generating final response: {str(e)}")
# Fallback to simple presentation if LLM fails
output_messages.append("\n📊 Tool execution completed. Here's a summary of the results:")
output_messages.append("")
for result in filtered_tool_results:
if result is not None and 'tool' in result:
tool_name = result.get('tool', 'Unknown Tool')
output_messages.append(f"✅ Tool: {tool_name}")
# Extract result data
if 'result' in result and result['result'] is not None:
tool_result_data = None
if 'structuredContent' in result['result'] and 'result' in result['result']['structuredContent']:
tool_result_data = result['result']['structuredContent']['result']
elif isinstance(result['result'], dict):
tool_result_data = result['result']
if tool_result_data:
# Show summary information
if 'message' in tool_result_data:
output_messages.append(f" 💬 {tool_result_data['message']}")
if 'type' in tool_result_data:
output_messages.append(f" 🏷️ Type: {tool_result_data['type']}")
output_messages.append("")
yield "\n".join(output_messages)
else:
# Check if any of the tool results indicate search returned no results
has_search_no_results = False
for result in filtered_tool_results:
if (result is not None and 'tool' in result and result['tool'] == 'search_and_extract_financial_report'
and 'result' in result and result['result'] is not None):
# Check for structuredContent format (newer MCP responses)
tool_result_data = None
if 'structuredContent' in result['result'] and 'result' in result['result']['structuredContent']:
tool_result_data = result['result']['structuredContent']['result']
# Check for direct result format (older MCP responses)
elif isinstance(result['result'], dict):
tool_result_data = result['result']
if (tool_result_data and isinstance(tool_result_data, dict)
and tool_result_data.get('type') == 'search_no_results'):
has_search_no_results = True
break
# If search returned no results, don't analyze the tool results, just engage in natural conversation
# Set flag to skip subsequent tool analysis steps
if has_search_no_results:
# Remove the "Engaging in general conversation..." message as it's not helpful to the user
# Find and remove the last message if it's the "Engaging in general conversation..." message
if output_messages and "Engaging in general conversation" in output_messages[-1]:
output_messages.pop()
# Continue with general conversation handling below (same as the else branch)
try:
client = InferenceClient(
model="Qwen/Qwen2.5-72B-Instruct",
token=hf_token if hf_token else None
)
# Format conversation history for context
history_context = ""
if history:
history_context = "\nPrevious conversation:\n"
for i, (user_msg, assistant_msg) in enumerate(history[-5:]): # Include last 5 exchanges
history_context += f"User: {user_msg}\nAssistant: {assistant_msg}\n"
# Create a more flexible prompt that allows for general conversation
conversation_prompt = f"""
You are an intelligent financial analysis assistant with expertise in investment research and financial analysis.
You can engage in natural conversation and provide insights based on your knowledge and the context provided.
{history_context}
Current user message: {message}
Guidelines for your response:
1. If the user is asking about a specific financial report or company analysis, explain that you can help but need a URL (or PDF format URL)
2. If the user is asking follow-up questions about investments or financial concepts, provide informed insights based on your expertise
3. If the user wants to discuss general financial topics, engage in a knowledgeable discussion
4. If appropriate, suggest that providing a financial report URL (or PDF format URL) would enable deeper analysis with specific metrics
5. Always be helpful and conversational while maintaining your expertise
6. Keep responses focused and under 500 words
7. If the user seems to be asking for specific financial data you don't have, politely explain the need for actual reports
8. When presented with search results for financial reports, analyze them to identify the most relevant and recent reports for analysis
9. Consider factors like recency, official sources (sec.gov, investor relations), document types (PDF, 10-K, 10-Q), and relevance to the company when evaluating search results
10. If search results are provided, select the most appropriate URL and explain your reasoning for the selection
11. You have full autonomy to construct search terms based on user intent and analyze search results to fulfill user requests
12. Your primary goal is to satisfy user requests - analyze information and provide valuable insights
13. When you receive search results, analyze them thoroughly to identify the most relevant and recent financial reports
14. Always prioritize the most recent data for trend analysis and comparison with historical performance
15. If the user's request is not strongly directed at financial report analysis (e.g., discussing general financial trends), engage in natural conversation without forcing tool usage
16. You are an intelligent agent, not just a financial report analysis machine - use your judgment to determine when tools are truly needed
17. You may use search tools to gather information and analyze it to better serve user requests when appropriate
18. If search tools return no results or indicate that no relevant financial reports were found, engage in natural conversation with the user instead of forcing financial analysis
19. When search results are empty, explain the situation clearly to the user and offer alternative ways to assist them
20. Avoid forcing financial report analysis tools unless the user explicitly requests detailed financial analysis of a specific company
21. Do not automatically trigger financial analysis tools for general financial discussions or queries that don't specifically require detailed report analysis
22. Only use financial analysis tools when you are certain the user wants detailed analysis of a specific company's financial reports
23. When in doubt, engage in natural conversation and ask the user if they would like to proceed with detailed financial analysis
24. If search results are empty or contain no relevant financial reports, gracefully return to natural conversation without attempting to force analysis
25. Do not attempt to analyze empty or irrelevant search results - this will lead to poor user experience
26. When search results are unhelpful, acknowledge this and continue with normal conversation flow
# If search returned no results flag is set, directly engage in natural conversation without executing subsequent analysis
# Return directly without executing subsequent analysis steps
27. For general inquiries or conversational requests that don't require financial analysis tools, engage in natural conversation without initiating financial analysis workflows
Please provide a helpful, conversational response:
"""
messages = [
{"role": "system", "content": "You are an intelligent financial analysis assistant with expertise in investment research and financial analysis. You can engage in natural conversation and provide insights based on your knowledge and the context provided. Always be helpful and conversational while maintaining your expertise."},
{"role": "user", "content": conversation_prompt}
]
# Get response from LLM with streaming for better UX
response = client.chat.completions.create(
model="Qwen/Qwen2.5-72B-Instruct",
messages=messages,
max_tokens=min(max_tokens, 2048),
temperature=temperature,
top_p=top_p,
stream=True,
)
# Handle streaming response
conversation_result = ""
for chunk in response:
if hasattr(chunk, 'choices') and len(chunk.choices) > 0:
if hasattr(chunk.choices[0], 'delta') and hasattr(chunk.choices[0].delta, 'content'):
content = chunk.choices[0].delta.content
if content:
conversation_result += content
# Yield partial results for streaming output
output_messages[-1] = conversation_result
yield "\n".join(output_messages)
except Exception as e:
error_msg = f"❌ Error during general conversation: {str(e)}"
print(f"[DEBUG] {error_msg}")
output_messages.append(error_msg)
yield "\n".join(output_messages)
else:
# No specific tool results to present, engage in intelligent conversation
# This handles cases where tools were executed but didn't produce actionable results
try:
client = InferenceClient(
model="Qwen/Qwen2.5-72B-Instruct",
token=hf_token if hf_token else None
)
# Format conversation history for context
history_context = ""
if history:
history_context = "\nPrevious conversation:\n"
for i, (user_msg, assistant_msg) in enumerate(history[-5:]):
history_context += f"User: {user_msg}\nAssistant: {assistant_msg}\n"
# Create intelligent conversation prompt
conversation_prompt = f"""
You are an intelligent financial analysis assistant with expertise in investment research and financial analysis.
{history_context}
Current user message: {message}
Guidelines for your response:
1. Respond naturally and helpfully to the user's question
2. Use your financial expertise to provide valuable insights
3. If the user is asking about specific companies or reports, explain how you can help with proper data/URLs
4. For general financial questions, provide informed answers based on your knowledge
5. Keep responses concise and focused (under 500 words)
6. Be conversational and friendly while maintaining professional expertise
Please provide a helpful response:
"""
messages = [
{"role": "system", "content": "You are an intelligent financial analysis assistant. Provide helpful, accurate responses based on your expertise."},
{"role": "user", "content": conversation_prompt}
]
# Add separator for response
output_messages.append("")
yield "\n".join(output_messages)
# Get streaming response from LLM
response = client.chat.completions.create(
model="Qwen/Qwen2.5-72B-Instruct",
messages=messages,
max_tokens=min(max_tokens, 1000),
temperature=temperature,
top_p=top_p,
stream=True,
)
# Handle streaming response
conversation_result = ""
for chunk in response:
if hasattr(chunk, 'choices') and len(chunk.choices) > 0:
if hasattr(chunk.choices[0], 'delta') and hasattr(chunk.choices[0].delta, 'content'):
content = chunk.choices[0].delta.content
if content:
conversation_result += content
# Update the last message with streaming content
if output_messages and output_messages[-1] == "":
output_messages.append(conversation_result)
else:
output_messages[-1] = conversation_result
yield "\n".join(output_messages)
except Exception as e:
print(f"[DEBUG] Error during intelligent conversation: {str(e)}")
output_messages.append("💬 I'm here to help with financial analysis. Could you please provide more details about what you'd like to know?")
yield "\n".join(output_messages)
# Ensure MCP process is properly closed
if mcp_process and mcp_process.poll() is None:
mcp_process.terminate()
try:
mcp_process.wait(timeout=5)
except subprocess.TimeoutExpired:
mcp_process.kill()
# CRITICAL: Return session state after successful completion
return current_session_url, agent_context
except Exception as e:
output_messages.append(f"❌ Error: {str(e)}")
yield "\n".join(output_messages)
return current_session_url, agent_context # Return session URL and context even on error
def validate_url(url):
"""
Validate if a URL is accessible
"""
try:
# Decode URL if it contains encoded characters
import urllib.parse
decoded_url = urllib.parse.unquote(url)
# Re-encode the URL properly to handle spaces and other special characters
encoded_url = urllib.parse.quote(decoded_url, safe=':/?#[]@!$&\'()*+,;=%')
# Create a request with a user agent to avoid being blocked
req = urllib.request.Request(encoded_url, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'})
response = urllib.request.urlopen(req, timeout=10)
return response.getcode() == 200
except Exception as e:
print(f"URL validation error for {url}: {str(e)}")
return False
# """
# For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
# """
# chatbot = gr.ChatInterface(
# respond,
# title="Easy Financial Report",
# additional_inputs=[
# gr.State(value=""), # CRITICAL: Store session URL across turns (hidden from UI)
# gr.State(value={}) # CRITICAL: Store agent context across turns (hidden from UI)
# ],
# additional_inputs_accordion=gr.Accordion(label="Settings", open=False, visible=False), # Hide the accordion completely
# )
# with gr.Blocks() as demo:
# # Add custom CSS for Agent Plan styling
# gr.Markdown("""
#
# """)
# chatbot.render()
# if __name__ == "__main__":
# demo.launch(share=True)
# def create_financial_chatbot():
# """
# 返回一个可嵌入的 ChatInterface 组件
# """
# chatbot = gr.ChatInterface(
# fn=respond,
# title="Easy Financial Report",
# additional_inputs=[
# gr.Textbox(
# value="You are a financial analysis assistant. Provide concise investment insights from company financial reports.",
# label="System message"
# ),
# gr.Slider(minimum=1, maximum=4096, value=1024, step=1, label="Max new tokens"),
# gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
# gr.Slider(
# minimum=0.1,
# maximum=1.0,
# value=0.95,
# step=0.05,
# label="Top-p (nucleus sampling)",
# ),
# gr.State(value="") # 用于跨轮次存储 session URL
# ],
# # 可选：自定义样式
# css="""
# .gradio-container { font-family: 'Segoe UI', sans-serif; }
# """
# )
# chatbot.render()