Building a Personal Finance Advisor with LangGraph and MCP

Use Case Overview

Objective: Create an AI agent that helps users manage finances by integrating banking APIs, investment platforms, and budgeting tools. The agent provides personalized advice, tracks spending, and suggests financial strategies.

Key Features:

• Real-time account balance checks.
• Spending pattern analysis.
• Investment recommendations based on risk tolerance.
• Budget tracking and alerts.

Core Components

LangGraph

• Role: Manages conversation states, remembers user preferences, and orchestrates multi-step workflows (e.g., “Analyze my spending → Suggest a budget → Recommend investments”).

StateGraph: Tracks:

• User’s financial goals (e.g., “Save $10k for a vacation”).
• Transaction history.
• Active tools (e.g., budget alerts).

MCP (Model Context Protocol)

Dynamic Tool Discovery: Connects to services like:

• Banking APIs (e.g., Plaid, Stripe).
• Investment platforms (e.g., Robinhood, Coinbase).
• Budgeting tools (e.g., Mint, YNAB).

Step-by-Step Implementation

Step 1: Imports

import asyncio
from typing import Optional, List, Dict
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import AIMessage, ToolMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder, SystemMessagePromptTemplate, HumanMessagePromptTemplate
from langchain_openai import ChatOpenAI
from mcp.server.fastmcp import FastMCP
from mcp.client import MultiServerMCPClient
from pydantic import BaseModel, Field
import dotenv
import os

# Load environment variables
dotenv.load_dotenv()

Step 2: Register Financial Tools via MCP

mcp = FastMCP()

class AccountBalanceQuery(BaseModel):
    bank_id: str = Field(..., description="Bank account ID")

@mcp.tool()
async def get_account_balance(query: AccountBalanceQuery) -> float:
    """Fetch real-time balance from banking API"""
    # Mock implementation - replace with actual API call
    return 4523.18  # Example balance

class SpendingAnalysisQuery(BaseModel):
    timeframe: str = Field(..., description="e.g., 'last_month' or 'last_30_days'")

@mcp.tool()
def analyze_spending(query: SpendingAnalysisQuery) -> dict:
    """Categorize spending patterns"""
    # Mock implementation
    return {
        "dining": 450.00,
        "groceries": 300.00,
        "entertainment": 150.00
    }

class InvestmentRecommendationQuery(BaseModel):
    risk_tolerance: str = Field(..., description="low/medium/high")

@mcp.tool()
def suggest_investments(query: InvestmentRecommendationQuery) -> List[str]:
    """Recommend investment products"""
    return ["VTI (Total Stock Market ETF)", "BND (Bond ETF)"] if query.risk_tolerance == "medium" else ["SPY (S&P 500)"]

Step 3: Graph Nodes

def get_system_prompt():
    return """You are a financial advisor AI. Available tools:
    - get_account_balance: Check bank balances
    - analyze_spending: Categorize expenses
    - suggest_investments: Recommend portfolios
    Always explain numbers in simple terms."""

def create_chatbot():
    prompt = ChatPromptTemplate.from_messages([
        SystemMessagePromptTemplate.from_template(get_system_prompt()),
        MessagesPlaceholder(variable_name="messages")
    ])
    llm = ChatOpenAI(model="gpt-4o-mini")
    return prompt | llm

async def process_query(state: MessagesState):
    chain = create_chatbot()
    response = await chain.ainvoke(state)
    return {"messages": state["messages"] + [response]}

async def execute_tools(state: MessagesState):
    messages = state["messages"]
    last_msg = messages[-1]
    
    tool_calls = getattr(last_msg, "tool_calls", [])
    if not tool_calls:
        return {"messages": messages}

    results = []
    async with MultiServerMCPClient(
        {"server": {"url": "http://localhost:8000/sse", "transport": "sse"}}
    ) as client:
        tools = await client.get_tools()
        for call in tool_calls:
            tool = next((t for t in tools if t.name == call.name), None)
            if tool:
                try:
                    result = await tool(**call.args) if asyncio.iscoroutinefunction(tool) else tool(**call.args)
                    results.append(ToolMessage(
                        content=str(result),
                        tool_call_id=getattr(call, "id", "0"),
                        name=call.name
                    ))
                except Exception as e:
                    results.append(AIMessage(content=f"Tool error: {str(e)}"))
    
    return {"messages": messages + results}

Step 4: Workflow Construction

def route_tool_calls(state: MessagesState):
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "execute_tools"
    return "end"

workflow = StateGraph(MessagesState)
workflow.add_node("chatbot", process_query)
workflow.add_node("tools", execute_tools)
workflow.add_edge(START, "chatbot")
workflow.add_conditional_edges(
    "chatbot",
    route_tool_calls,
    {"execute_tools": "tools", "end": END}
)
workflow.add_edge("tools", "chatbot")
agent = workflow.compile()

Step 5: Execution

async def main():
    # Start MCP server in background
    server_task = asyncio.create_task(mcp.run(transport="sse"))
    
    # Run agent
    query = "What's my current bank balance and spending pattern?"
    result = await agent.ainvoke({
        "messages": [HumanMessage(content=query)]
    })
    
    for msg in result["messages"]:
        print(f"{type(msg).__name__}: {msg.content}")
    
    await server_task  # Cleanup

if __name__ == "__main__":
    asyncio.run(main())

Example Interaction

User: “How much did I spend on dining last month?”

1. LangGraph routes the query to process_query.
2. The agent detects the need for analyze_spending("last_month").
3. MCP executes the tool and appends results:

{"Dining": "$450", "Groceries": "$300", "Entertainment": "$150"}

The agent responds: “You spent $450 on dining last month. Want to set a budget for this month?”

Advanced Features

• Budget Alerts: Use LangGraph to trigger notifications when spending exceeds limits.
• Investment Simulations: Integrate tools like simulate_portfolio(risk_level) to show potential returns.
• Debt Management: Add tools to optimize debt repayment strategies.

Challenges & Solutions

• Data privacy concerns : Use OAuth2 and end-to-end encryption.
• Real-time API latency : Cache frequent queries (e.g., balances).
• Complex financial jargon : Add a node to simplify explanations.

Conclusion

This use case demonstrates how LangGraph and MCP enable:

• Adaptability: Integrate new financial services without code changes.
• Personalisation: Maintain context across interactions (e.g., tracking goals).
• Security: Handle sensitive data securely through MCP’s protocol.