mcpField
A
agno

agno

github.com/agno-agiAgent 工作流26387 次安装36 周前更新
代码仓库

关于 agno

Agno Skill 为使用 Agno 构建 AI Agent 提供指导,适用于开发单 Agent、多 Agent 团队和工作流的开发者。内容涵盖工具、记忆、持久化及 MCP 服务集成。

主要功能

  1. 01配置带工具的 Agent
  2. 02构建多 Agent 团队
  3. 03编排多种工作流步骤
  4. 04实现会话持久化与记忆
  5. 05集成多个 MCP 服务

适用场景

  1. 01构建带结构化输出的 Agent
  2. 02按角色委派任务
  3. 03部署 Agent API
  4. 04连接 stdio、SSE 或 HTTP MCP 服务

MCP 相关

指导通过 stdio、SSE 和 Streamable HTTP 集成 MCP 服务,并提供 MCPTools 与 MultiMCPTools 的用法。示例包括将 MCP 工具接入 Agent。

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/agno-agi/agno-skills --skill agno

安装后在 Claude Code 中输入/agno确认已加载。

包含的文件

除 SKILL.md 外,安装时一起下载的参考资料和脚本。

references/
agents.md
learning.md
mcp.md
models.md
teams.md
tools.md
workflows.md

详细介绍

Agno 技能

使用 Agno 构建可用于生产环境的 AI Agent。Agno 是一个轻量、与模型无关的框架,支持 Agent、团队、工作流和 MCP 集成。

何时使用此技能

出现以下情况时,应触发此技能:

  • 使用工具、记忆、结构化输出或知识构建 AI Agent
  • 创建基于角色委派任务的多 Agent 团队
  • 实现包含顺序、并行、条件或路由步骤的工作流
  • 集成 MCP 服务(stdio、SSE 或 Streamable HTTP)
  • 使用 AgentOS(基于 FastAPI 的运行时)部署 Agent
  • 使用 LearningMachine(用户档案、实体记忆、会话上下文)
  • 调试 Agent 行为或优化性能

架构概览

Agent          - Single autonomous AI unit (model + tools + instructions)
Team           - Multiple agents coordinated by a leader (route/broadcast/tasks modes)
Workflow       - Pipeline-based execution (Step, Parallel, Condition, Loop, Router)
AgentOS        - FastAPI runtime for deploying agents as production APIs
LearningMachine - Persistent learning across sessions (profiles, memory, knowledge)

快速参考

1. 使用工具的基础 Agent
from agno.agent import Agent
from agno.models.google import Gemini
from agno.tools.yfinance import YFinanceTools

agent = Agent(
    name="Finance Agent",
    model=Gemini(id="gemini-3-flash-preview"),
    tools=[YFinanceTools()],
    add_datetime_to_context=True,
    markdown=True,
)

agent.print_response("Give me a quick brief on NVIDIA", stream=True)
2. 使用 Pydantic 的结构化输出
from typing import List, Optional
from agno.agent import Agent
from agno.models.google import Gemini
from agno.tools.yfinance import YFinanceTools
from pydantic import BaseModel, Field

class StockAnalysis(BaseModel):
    ticker: str = Field(..., description="Stock ticker symbol")
    company_name: str = Field(..., description="Full company name")
    current_price: float = Field(..., description="Current price in USD")
    summary: str = Field(..., description="One-line summary")
    key_drivers: List[str] = Field(..., description="2-3 key growth drivers")
    recommendation: str = Field(..., description="Buy, Hold, or Sell")

agent = Agent(
    model=Gemini(id="gemini-3-flash-preview"),
    tools=[YFinanceTools()],
    output_schema=StockAnalysis,
)

response = agent.run("Analyze NVIDIA")
analysis: StockAnalysis = response.content
print(f"{analysis.company_name}: {analysis.recommendation}")
3. 使用存储的 Agent(会话持久化)
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.google import Gemini

agent = Agent(
    model=Gemini(id="gemini-3-flash-preview"),
    db=SqliteDb(db_file="tmp/agents.db"),
    add_history_to_context=True,
    num_history_runs=5,
    markdown=True,
)

# Same session_id = continuous conversation across runs
agent.print_response("Analyze NVDA", session_id="my-session", stream=True)
agent.print_response("Compare that to Tesla", session_id="my-session", stream=True)
4. 使用记忆的 Agent(用户偏好)
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.memory import MemoryManager
from agno.models.google import Gemini

db = SqliteDb(db_file="tmp/agents.db")

agent = Agent(
    model=Gemini(id="gemini-3-flash-preview"),
    db=db,
    memory_manager=MemoryManager(
        model=Gemini(id="gemini-3-flash-preview"),
        db=db,
    ),
    enable_agentic_memory=True,  # Agent decides when to store/recall
    markdown=True,
)

# Agent remembers user preferences across sessions
agent.print_response(
    "I'm interested in AI stocks. My risk tolerance is moderate.",
    user_id="[email protected]",
    stream=True,
)
5. 多 Agent 团队
from agno.agent import Agent
from agno.models.google import Gemini
from agno.team.team import Team
from agno.tools.yfinance import YFinanceTools

bull = Agent(
    name="Bull Analyst",
    role="Make the investment case FOR a stock",
    model=Gemini(id="gemini-3-flash-preview"),
    tools=[YFinanceTools()],
)

bear = Agent(
    name="Bear Analyst",
    role="Make the investment case AGAINST a stock",
    model=Gemini(id="gemini-3-flash-preview"),
    tools=[YFinanceTools()],
)

team = Team(
    name="Investment Research",
    model=Gemini(id="gemini-3-flash-preview"),
    members=[bull, bear],
    instructions=["Get both perspectives, then synthesize a balanced recommendation"],
    show_members_responses=True,
    markdown=True,
)

team.print_response("Should I invest in NVIDIA?", stream=True)
6. 顺序工作流
from agno.agent import Agent
from agno.models.google import Gemini
from agno.tools.yfinance import YFinanceTools
from agno.workflow import Step, Workflow

data_agent = Agent(name="Data Gatherer", model=Gemini(id="gemini-3-flash-preview"), tools=[YFinanceTools()])
analyst = Agent(name="Analyst", model=Gemini(id="gemini-3-flash-preview"))
writer = Agent(name="Report Writer", model=Gemini(id="gemini-3-flash-preview"), markdown=True)

workflow = Workflow(
    name="Research Pipeline",
    steps=[
        Step(name="Gather Data", agent=data_agent),
        Step(name="Analyze", agent=analyst),
        Step(name="Write Report", agent=writer),
    ],
)

workflow.print_response("Analyze NVIDIA for investment", stream=True)
7. MCP 服务集成(stdio)
import asyncio
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.mcp import MCPTools

async def run_agent(message: str) -> None:
    async with MCPTools(command="uvx mcp-server-git") as mcp_tools:
        agent = Agent(model=Claude(id="claude-sonnet-4-5-20250929"), tools=[mcp_tools])
        await agent.aprint_response(message, stream=True)

asyncio.run(run_agent("What is the license for this project?"))
8. MCP 服务(Streamable HTTP)
import asyncio
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.mcp import MCPTools

async def run_agent(message: str) -> None:
    async with MCPTools(
        transport="streamable-http",
        url="https://docs.agno.com/mcp",
    ) as mcp_tools:
        agent = Agent(model=Claude(id="claude-sonnet-4-5-20250929"), tools=[mcp_tools], markdown=True)
        await agent.aprint_response(message, stream=True)

asyncio.run(run_agent("What is Agno?"))
9. 多个 MCP 服务
import asyncio
from os import getenv
from agno.agent import Agent
from agno.tools.mcp import MultiMCPTools

async def run_agent(message: str) -> None:
    mcp_tools = MultiMCPTools(
        commands=["npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt"],
        urls=["http://localhost:8000/mcp"],
        urls_transports=["streamable-http"],
        timeout_seconds=30,
    )
    await mcp_tools.connect()
    agent = Agent(tools=[mcp_tools], markdown=True)
    await agent.aprint_response(message, stream=True)
    await mcp_tools.close()

asyncio.run(run_agent("Find listings in Barcelona"))
10. LearningMachine(持久化学习)
from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.learn import LearningMachine, LearningMode, UserProfileConfig
from agno.models.openai import OpenAIResponses

db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    db=db,
    learning=LearningMachine(
        user_profile=UserProfileConfig(mode=LearningMode.ALWAYS),
    ),
    markdown=True,
)

agent.print_response("Hi! I'm Alice, call me Ali.", user_id="[email protected]", stream=True)
# Profile fields (name, preferred_name) captured automatically

关键模式

模式:MCP 连接生命周期

始终关闭 MCP 连接。使用异步上下文管理器或 try/finally:

# Preferred: context manager
async with MCPTools(command="uvx mcp-server-git") as tools:
    agent = Agent(tools=[tools])
    await agent.aprint_response("query")

# Alternative: manual lifecycle
tools = MCPTools(command="uvx mcp-server-git")
await tools.connect()
try:
    agent = Agent(tools=[tools])
    await agent.aprint_response("query")
finally:
    await tools.close()
模式:生产数据库(PostgreSQL)
from agno.db.postgres import PostgresDb
db = PostgresDb(db_url="postgresql+psycopg://user:pass@localhost:5432/agno")
agent = Agent(db=db, add_history_to_context=True)
模式:调试模式
agent = Agent(debug_mode=True)  # Detailed logs of messages, tools, tokens
模式:自定义工具
from agno.tools.decorator import tool

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"Weather in {city}: 72F, sunny"

agent = Agent(tools=[get_weather])

重要规则

  • 切勿在循环中创建 Agent——为提升性能,应重复使用 Agent
  • 对结构化响应使用 output_schema(不要使用自由格式解析)
  • 生产环境使用 PostgreSQL,SQLite 仅用于开发
  • 同步和异步均可用——所有公共方法都有对应的异步版本(以 a 为前缀)
  • 始终关闭 MCP 连接——使用 try/finally 或异步上下文管理器
  • 排查问题时启用 debug_mode=True

参考文件

详细文档见 references/:

  • agents.md - Agent 参数、配置、工具、记忆、知识、护栏
  • teams.md - 团队模式(route/broadcast/tasks)、成员协调
  • workflows.md - 步骤类型(Step、Parallel、Condition、Loop、Router)
  • mcp.md - MCP 集成(stdio、SSE、Streamable HTTP)、MultiMCPTools
  • tools.md - 内置工具列表、自定义工具创建、工具钩子
  • learning.md - LearningMachine 存储(profile、memory、session、knowledge、entity)
  • models.md - 支持的模型提供商和配置

资源

评论

支持 Markdown