How to Integrate FastAPI for wealth management with LangChain for AI agents

FastAPI gives you the HTTP surface for wealth-management workflows: portfolio lookup, suitability checks, KYC status, trade initiation, and client servicing. LangChain gives you the agent layer to decide which workflow to call, when to ask for clarification, and how to turn unstructured user requests into structured actions.

The useful pattern is simple: FastAPI exposes deterministic financial operations, while LangChain orchestrates them as tools inside an AI agent. That lets you build assistants that can answer client questions, retrieve account data, and route requests without hardcoding every conversation path.

Prerequisites

• •Python 3.10+
• •A FastAPI service already running for your wealth-management backend
• •langchain, langchain-openai, fastapi, uvicorn, and httpx installed
• •An OpenAI API key or another LangChain-compatible LLM provider
• •A wealth-management API with endpoints like:
  • •GET /clients/{client_id}/portfolio
  • •GET /clients/{client_id}/risk-profile
  • •POST /advice/summary
• •Basic familiarity with async Python and JSON APIs

Install the packages:

pip install fastapi uvicorn httpx langchain langchain-openai pydantic

Integration Steps

1) Define the FastAPI wealth-management endpoints

Start by exposing stable backend operations. Keep these endpoints narrow and deterministic so the agent can call them safely.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List

app = FastAPI(title="Wealth Management API")

class Holding(BaseModel):
    symbol: str
    quantity: float
    market_value: float

class PortfolioResponse(BaseModel):
    client_id: str
    total_value: float
    holdings: List[Holding]

@app.get("/clients/{client_id}/portfolio", response_model=PortfolioResponse)
async def get_portfolio(client_id: str):
    if client_id != "client_123":
        raise HTTPException(status_code=404, detail="Client not found")

    return PortfolioResponse(
        client_id=client_id,
        total_value=1250000.00,
        holdings=[
            Holding(symbol="AAPL", quantity=120, market_value=24000),
            Holding(symbol="MSFT", quantity=80, market_value=32000),
        ],
    )

@app.get("/clients/{client_id}/risk-profile")
async def get_risk_profile(client_id: str):
    return {"client_id": client_id, "risk_score": 72, "profile": "moderate"}

@app.post("/advice/summary")
async def advice_summary(payload: dict):
    return {
        "client_id": payload["client_id"],
        "summary": "Portfolio is concentrated in large-cap tech. Consider diversification into fixed income."
    }

This is the contract your agent will depend on. In production, add auth, audit logs, idempotency keys for write actions, and strict schema validation.

2) Wrap the FastAPI endpoints as LangChain tools

LangChain agents work best when each external capability is a tool with a clear input/output shape. Use httpx to call your FastAPI service from tool functions.

import httpx
from langchain_core.tools import tool

BASE_URL = "http://localhost:8000"

@tool
async def fetch_portfolio(client_id: str) -> dict:
    """Fetch a client's portfolio from the wealth management API."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{BASE_URL}/clients/{client_id}/portfolio")
        resp.raise_for_status()
        return resp.json()

@tool
async def fetch_risk_profile(client_id: str) -> dict:
    """Fetch a client's risk profile from the wealth management API."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{BASE_URL}/clients/{client_id}/risk-profile")
        resp.raise_for_status()
        return resp.json()

@tool
async def generate_advice_summary(client_id: str) -> dict:
    """Generate an advice summary using the wealth management API."""
    async with httpx.AsyncClient() as client:
        portfolio_resp = await client.get(f"{BASE_URL}/clients/{client_id}/portfolio")
        portfolio_resp.raise_for_status()

        payload = {"client_id": client_id, "portfolio": portfolio_resp.json()}
        resp = await client.post(f"{BASE_URL}/advice/summary", json=payload)
        resp.raise_for_status()
        return resp.json()

This keeps the LLM away from raw database access or business logic. The model only sees tools it can invoke.

3) Build a LangChain agent that uses those tools

Now connect the tools to an agent. For modern LangChain setups, use a chat model plus tool-calling support.

import asyncio
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a wealth management assistant. Use tools for factual account data."),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

tools = [fetch_portfolio, fetch_risk_profile, generate_advice_summary]

agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

async def run():
    result = await executor.ainvoke({
        "input": "Summarize client_123's portfolio and risk profile."
    })
    print(result["output"])

# asyncio.run(run())

The important part here is separation of concerns:

4) Add guardrails for financial workflows

For wealth-management systems, don’t let the agent freely improvise on trade or advice actions. Validate inputs before calling your backend and restrict what actions are exposed as tools.

from pydantic import BaseModel, Field

class ClientQuery(BaseModel):
    client_id: str = Field(min_length=3)
    request_type: str = Field(pattern="^(portfolio|risk_profile|advice_summary)$")

def validate_request(data: dict) -> ClientQuery:
    return ClientQuery(**data)

@tool
async def safe_client_lookup(payload: dict) -> dict:
    """Validated lookup for regulated workflows."""
    query = validate_request(payload)

    async with httpx.AsyncClient() as client:
        if query.request_type == "portfolio":
            resp = await client.get(f"{BASE_URL}/clients/{query.client_id}/portfolio")
        elif query.request_type == "risk_profile":
            resp = await client.get(f"{BASE_URL}/clients/{query.client_id}/risk-profile")
        else:
            port = await client.get(f"{BASE_URL}/clients/{query.client_id}/portfolio")
            resp = await client.post(
                f"{BASE_URL}/advice/summary",
                json={"client_id": query.client_id, "portfolio": port.json()},
            )

        resp.raise_for_status()
        return resp.json()

This pattern matters in regulated environments. You want explicit request types, schema validation, and auditability before any downstream action happens.

Testing the Integration

Run your FastAPI app first:

uvicorn main:app --reload --port 8000

Then test the LangChain side with a direct invocation:

import asyncio

async def test_agent():
    result = await executor.ainvoke({
        "input": "Get the portfolio for client_123 and summarize any concentration risk."
    })
    print(result["output"])

asyncio.run(test_agent())

Expected output:

Client client_123 has a $1.25M portfolio concentrated in large-cap tech.
Main holdings include AAPL and MSFT.
Risk profile is moderate; consider diversification into fixed income or non-correlated assets.

If you want a lower-level integration check without the agent layer:

import asyncio

async def test_tool():
    data = await fetch_portfolio.ainvoke("client_123")
    print(data["total_value"])
    
asyncio.run(test_tool())

Expected output:

1250000.0

Real-World Use Cases

• •Client servicing assistant that answers “What’s my current allocation?” by calling portfolio and risk endpoints through an agent.
• •Advisor copilot that drafts investment summaries after pulling holdings, risk score, and recent activity from your FastAPI backend.
• •Operations assistant that routes KYC status checks, suitability reviews, and account-data lookups without exposing internal services directly to users.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.