import httpx
from agno.agent import Agent
from agno.tools import tool
from typing import Any, Callable, Dict
def logger_hook(function_name: str, function_call: Callable, arguments: Dict[str, Any]):
"""包装工具执行的 Hook 函数"""
print(f"即将调用 {function_name},参数为:{arguments}")
result = function_call(**arguments)
print(f"函数调用完成,结果为:{result}")
return result
@tool(
name="fetch_hackernews_stories", # 工具的自定义名称(否则使用函数名称)
description="获取 Hacker News 的热门故事", # 自定义描述(否则使用函数文档字符串)
show_result=True, # 函数调用后显示结果
stop_after_tool_call=True, # 工具调用后立即返回结果并停止 Agent
tool_hooks=[logger_hook], # 在执行前后运行的 Hook
requires_confirmation=True, # 在执行前需要用户确认
cache_results=True, # 启用结果缓存
cache_dir="/tmp/agno_cache", # 自定义缓存目录
cache_ttl=3600 # 缓存的 TTL(以秒为单位,1 小时)
)
def get_top_hackernews_stories(num_stories: int = 5) -> str:
"""
从 Hacker News 获取热门故事。
Args:
num_stories: 要获取的故事数量(默认:5)
Returns:
str: 以文本格式的热门故事
"""
# 获取热门故事 ID
response = httpx.get("https://hacker-news.firebaseio.com/v0/topstories.json")
story_ids = response.json()
# 获取故事详情
stories = []
for story_id in story_ids[:num_stories]:
story_response = httpx.get(f"https://hacker-news.firebaseio.com/v0/item/{story_id}.json")
story = story_response.json()
stories.append(f"{story.get('title')} - {story.get('url', '无 URL')}")
return "\n".join(stories)
agent = Agent(tools=[get_top_hackernews_stories])
agent.print_response("给我看看 Hacker News 的热门新闻")