MCP в Cursor IDE: подключаем AI-агенту внешние данные без возни с API

{
  "mcpServers": {
    "google-trends": {
      "command": "python",
      "args": ["-m", "trends_server"],
      "env": {
        "SERPAPI_KEY": "your-key-here"
      }
    }
  }
}

{
  "mcpServers": {
    "google-trends": {
      "command": "python",
      "args": ["-m", "trends_server"]
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "POSTGRES_URL": "postgresql://user:pass@localhost:5432/mydb"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/docs"]
    }
  }
}

from mcp.server.fastmcp import FastMCP
from pytrends.request import TrendReq
import json
mcp = FastMCP("google-trends")
pytrends = TrendReq(hl="ru", tz=180)  # русский язык, московское время
@mcp.tool()
def get_interest_over_time(query: str, timeframe: str = "today 3-m") -> str:
    """Получить динамику интереса к запросу за период.
    Аргументы:
        query: поисковый запрос (например 'machine learning')
        timeframe: период ('today 3-m', 'today 12-m', '2024-01-01 2024-12-31')
    Возвращает JSON с данными по неделям.
    """
    pytrends.build_payload([query], timeframe=timeframe)
    df = pytrends.interest_over_time()
    if df.empty:
        return json.dumps({"error": f"Нет данных для '{query}'"})
    # Убираем колонку isPartial, она мешает анализу
    if "isPartial" in df.columns:
        df = df.drop(columns=["isPartial"])
    return df.to_json(date_format="iso")
@mcp.tool()
def get_related_queries(query: str) -> str:
    """Получить связанные запросы: топ (самые частые) и растущие (с наибольшим ростом).
    Аргументы:
        query: поисковый запрос
    Растущие запросы с пометкой 'Breakout' выросли более чем на 5000%.
    """
    pytrends.build_payload([query], timeframe="today 3-m")
    related = pytrends.related_queries()
    result = {}
    for key in related:
        rising = related[key].get("rising")
        top = related[key].get("top")
        result[key] = {
            "rising": rising.to_dict() if rising is not None else {},
            "top": top.to_dict() if top is not None else {},
        }
    return json.dumps(result, ensure_ascii=False, default=str)
@mcp.tool()
def compare_queries(queries: list[str], timeframe: str = "today 3-m") -> str:
    """Сравнить до 5 запросов по динамике интереса.
    Аргументы:
        queries: список запросов для сравнения (максимум 5)
        timeframe: период
    Значения нормализованы: 100 = пик интереса, остальные относительно него.
    """
    if len(queries) > 5:
        return json.dumps({"error": "Максимум 5 запросов"})
    pytrends.build_payload(queries, timeframe=timeframe)
    df = pytrends.interest_over_time()
    if "isPartial" in df.columns:
        df = df.drop(columns=["isPartial"])
    return df.to_json(date_format="iso")
@mcp.tool()
def get_trending_searches(country: str = "russia") -> str:
    """Получить сегодняшние трендовые поисковые запросы в стране.
    Аргументы:
        country: страна ('russia', 'united_states', 'germany')
    """
    trending = pytrends.trending_searches(pn=country)
    return trending.to_json(force_ascii=False)
if __name__ == "__main__":
    mcp.run(transport="stdio")

pip install mcp pytrends

from mcp.server.fastmcp import FastMCP
import asyncpg
import json
mcp = FastMCP("product-metrics")
DB_URL = "postgresql://readonly:pass@localhost:5432/analytics"
@mcp.tool()
async def get_daily_signups(days: int = 30) -> str:
    """Получить количество регистраций по дням за последние N дней."""
    conn = await asyncpg.connect(DB_URL)
    try:
        rows = await conn.fetch("""
            SELECT date_trunc('day', created_at) as day, count(*) as signups
            FROM users
            WHERE created_at > now() - interval '$1 days'
            GROUP BY 1 ORDER BY 1
        """, days)
        return json.dumps([{"day": str(r["day"].date()), "signups": r["signups"]} for r in rows])
    finally:
        await conn.close()
@mcp.tool()
async def get_feature_usage(feature_name: str, days: int = 7) -> str:
    """Получить статистику использования фичи: DAU, количество событий, среднее на пользователя."""
    conn = await asyncpg.connect(DB_URL)
    try:
        rows = await conn.fetch("""
            SELECT
                date_trunc('day', timestamp) as day,
                count(distinct user_id) as dau,
                count(*) as events,
                round(count(*)::numeric / nullif(count(distinct user_id), 0), 1) as avg_per_user
            FROM events
            WHERE feature = $1 AND timestamp > now() - interval '$2 days'
            GROUP BY 1 ORDER BY 1
        """, feature_name, days)
        return json.dumps([dict(r) for r in rows], default=str)
    finally:
        await conn.close()
@mcp.tool()
async def get_retention(cohort_month: str) -> str:
    """Получить retention когорты по месяцам. cohort_month в формате '2026-01'."""
    conn = await asyncpg.connect(DB_URL)
    try:
        rows = await conn.fetch("""
            SELECT
                months_since_signup,
                count(distinct user_id) as active_users,
                round(100.0 * count(distinct user_id) /
                    first_value(count(distinct user_id)) over (order by months_since_signup), 1) as retention_pct
            FROM user_activity_monthly
            WHERE cohort = $1
            GROUP BY 1 ORDER BY 1
        """, cohort_month)
        return json.dumps([dict(r) for r in rows], default=str)
    finally:
        await conn.close()
if __name__ == "__main__":
    mcp.run(transport="stdio")

{
  "mcpServers": {
    "product-metrics": {
      "command": "python",
      "args": ["-m", "metrics_server"],
      "env": {
        "DATABASE_URL": "postgresql://readonly:pass@localhost:5432/analytics"
      }
    }
  }
}

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
# Инструменты (те же, что в MCP-сервере, но как LangChain tools)
@tool
def get_trends(query: str) -> str:
    """Получить данные Google Trends."""
    pytrends.build_payload([query], timeframe="today 3-m")
    return pytrends.interest_over_time().to_json()
@tool
def get_related(query: str) -> str:
    """Получить связанные запросы."""
    pytrends.build_payload([query], timeframe="today 3-m")
    related = pytrends.related_queries()
    rising = related[query].get("rising")
    return rising.to_json() if rising is not None else "{}"
@tool
def compare(queries: list[str]) -> str:
    """Сравнить до 5 запросов."""
    pytrends.build_payload(queries[:5], timeframe="today 3-m")
    return pytrends.interest_over_time().to_json()
tools = [get_trends, get_related, compare]
# Состояние графа
class State(TypedDict):
    messages: list
    iteration: int
# LLM с инструментами
llm = ChatAnthropic(model="claude-sonnet-4-20250514").bind_tools(tools)
def analyst(state: State) -> dict:
    """Узел-аналитик: решает, какой инструмент вызвать."""
    response = llm.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}
def should_continue(state: State) -> str:
    """Условное ребро: есть ли вызовы инструментов?"""
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return "end"
# Сборка графа
workflow = StateGraph(State)
workflow.add_node("analyst", analyst)
workflow.add_node("tools", ToolNode(tools))
workflow.add_edge(START, "analyst")
workflow.add_conditional_edges("analyst", should_continue, {"tools": "tools", "end": END})
workflow.add_edge("tools", "analyst")  # после вызова инструмента — обратно к аналитику
agent = workflow.compile()
# Запуск
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "Найди растущие темы в области 'AI coding' и сравни топ-3"
    }],
    "iteration": 0,
})