dbt-mcp

import asyncio
from pathlib import Path

from crewai import Agent, LLM, Task, Crew
from crewai.mcp import MCPServerStdio


async def create_mcp_server(dbt_mcp_dir: str) -> MCPServerStdio:
    return MCPServerStdio(
        command="uvx",
        args=["--env-file", f"{dbt_mcp_dir}/.env", "dbt-mcp"],
        tool_filter=None,
        cache_tools_list=True,
    )


async def main():
    dbt_mcp_dir = Path(__file__).parent.parent.parent
    llm = LLM(model="gpt-4.1")
    server = await create_mcp_server(dbt_mcp_dir=dbt_mcp_dir)

    dbt_agent = Agent(
        role="dbt agent",
        goal="Ask the user for a question before taking any action.",
        backstory=(
            "You are a senior analytics engineer with deep expertise in dbt projects. "
            "You understand dbt models, sources, macros, tests, documentation, and "
            "project structure, and you know when to inspect the repository using tools "
            "instead of guessing."
        ),
        mcps=[server],
        llm=llm,
    )

    # Create a task
    dbt_task = Task(
        description="Answer the user question related to dbt using the available tools.",
        expected_output="A clear answer to user's question obtained via the tools.",
        agent=dbt_agent,
        human_input=True,
    )
    analysis_crew = Crew(
        agents=[dbt_agent], tasks=[dbt_task], memory=True, verbose=False
    )

    _results = await analysis_crew.kickoff_async()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Execution cancelled by user.")
    except Exception as e:
        print(f"Error occurred: {e}")