Retrieving Structured Output From MCP-Integrated LangGraph Agent

Author(s): Ruiwen (Rei-1)

Originally published on Towards AI.

Structured output transforms LLM-based applications and agentic systems into reliable, interoperable components by enforcing a clear, machine-readable schema (e.g., JSON with explicitly defined fields). This makes it easy for downstream systems — dashboards, databases, APIs, or UI components — to automatically ingest, validate, and act on model results without brittle parsing. At the same time, structured schemas enable rigorous testing, monitoring, and compliance audits. By adopting a shared schema, teams can confidently chain specialized models, write unit tests against expected fields, and build rich user experiences (charts, cards, conditional logic) from AI outputs. In regulated or complex environments, structured responses also provide the traceability and error‐handling guarantees that freeform text simply cannot match.

Although numerous tutorials demonstrate how to connect MCP servers with LangGraph agents, few explain how to extract and validate structured outputs once your agent is live. Most guides show you how to invoke MCP tools or illustrate a basic request/response flow — but then leave you to figure out how to enforce schemas or parse Pydantic results. As a result, developers often resort to manually inspecting raw text responses or cobbling together ad-hoc parsing logic — workarounds that undermine the reliability and automation gains MCP integration is meant to provide.

This lack of clear guidance can create a last-mile hurdle: without a step-by-step reference for retrieving and validating JSON objects that conform to predefined models, you can’t be certain your LangGraph + MCP workflows will produce predictable, machine-readable results that downstream systems can programmatically consume. By filling this gap, you’ll preserve the very benefits you set out to achieve — robust error handling, consistent data contracts, and seamless interoperability.

In this article, you will build an end-to-end pipeline for structured output retrieval. We’ll start by implementing an MCP server. Next, you’ll configure a LangGraph-powered client to communicate with that server, dynamically discover its tools, and invoke them asynchronously. Then, you’ll wire those tools into a robust AI agent using LangGraph’s ReAct framework — complete with custom prompts and Pydantic-based output parsing. Finally, we’ll show you how to retrieve, validate, and programmatically consume structured JSON responses, ensuring your agent’s decisions remain transparent, auditable, and reliable.

Setting up the MCP server

Let’s start to build a local weather MCP server adapted from Anthropic’s “For Server Developers” tutorial. This server implements theget_forecasttool which fetch weather information via API calls. A local MCP server operates on the same machine as the MCP client (the AI application). Communication between the client and the server typically occurs through standard input and output (stdio), allowing for direct and efficient data exchange. Save the following code as weather_server.py in your project directory.

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather") # Initilize the server
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

async def make_nws_request(url: str) -> dict[str, Any] | None:
 """Make a request to the NWS API with proper error handling."""
 headers = {
 "User-Agent": USER_AGENT,
 "Accept": "application/geo+json"
 }
 async with httpx.AsyncClient() as client:
 try:
 response = await client.get(url, headers=headers, timeout=30.0)
 response.raise_for_status()
 return response.json()
 except Exception:
 return None

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
 """Get weather forecast for a location.
 Args:
 latitude: Latitude of the location
 longitude: Longitude of the location
 """
 # First get the forecast grid endpoint
 points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
 points_data = await make_nws_request(points_url)
 if not points_data:
 return "Unable to fetch forecast data for this location."

 # Get the forecast URL from the points response
 forecast_url = points_data["properties"]["forecast"]
 forecast_data = await make_nws_request(forecast_url)
 if not forecast_data:
 return "Unable to fetch detailed forecast."

 return forecast_data

if __name__ == "__main__":
 mcp.run(transport='stdio')

You can also leverage remote MCP services running on external servers, which your clients access over standard web protocols. Instead of spawning a local process, your client points at an HTTP(S) endpoint and uses Server-Sent Events (SSE) or WebSockets for bidirectional streaming. Under the hood, the client discovers available tools and invokes them just as if they were local functions — only now the code and data live in the cloud.

Remote MCP deployments unlock a host of benefits: you can centralize maintenance and versioning of your toolset, autoscale the service independently of client workloads, and enforce security policies (authentication, auditing, rate limits) at the server boundary. They also let your agents tap into specialized cloud resources or third-party APIs — whether that’s high-resolution satellite imagery, large-scale language models, or proprietary financial data — without bloating the client. Popular hosts include Anthropic’s managed MCP platform, Smithery AI, and DIY deployments on AWS Lambda (behind API Gateway), Google Cloud Run, or Azure Functions, giving you full control over latency, compliance, and compute scaling.

LangGraph Agent + MCP servers

The code below illustrate how to set up an MCP client with a LangGraph agent. The client handles communication with the weather MCP server to access its tool. The agent then uses the ReAct loop — a cycle of “Reason → Act → Observe” — to process user inputs: it first generates an internal reasoning step (the “Thought”), chooses an appropriate tool to call, invokes that tool, observe the result, and repeats until it produces a final answer. Throughout this process, ReAct maintains a “scratchpad” of past thoughts, actions and observations so that each decision is contextually grounded.

# save to file agent.py
from dotenv import load_dotenv
import os
load_dotenv()
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

## server parameters
server_params={
 "weather": {
 "command": "python",
 "args": ["/absolute_path_to_your_server_file/weather_server.py"],
 "transport": "stdio",
 }
 }

async def main(query: str):
 async with MultiServerMCPClient(server_params) as client:
        tools = client.get_tools()
 agent = create_react_agent(model, tools)
 response = await agent.ainvoke({"messages": query})

 return response

# Run the async main function - queries for testing
query = "Forecast the weather in Raleigh, NC next week."

if __name__ == "__main__":
 response=asyncio.run(main(query))
 print('------------')
 last_message = response["messages"][-1]
 print(last_message.content)

Within this framework, we can easily register multiple MCP servers, both local and remote, by adding their configurations as new entries in the server_params dictionary. The weatherserver runs locally, whereas the sequential-thinkingserver is hosted remotely by smithery.ai.

## server parameters
server_params={
 "weather": {
 "command": "python",
 "args": ["/absolute_path_to_your_server_file/weather_server.py"],
 "transport": "stdio",
 },
 "sequential-thinking": {
 "command": "npx",
 "args": [
 "-y",
 "@smithery/cli@latest",
 "run",
 "@smithery-ai/server-sequential-thinking",
 "--key",
 "ef89e1ce-22ed-46ca-b0c1-52cc148b3521"
 ],
 "transport": "stdio"
 }
 }

To run the agent, simply execute the following command:

python agent.py

Here’s the result from running the query: “Forecast the weather in Raleigh, NC next week.”

Here's the weather forecast for Raleigh, NC next week:

- **Saturday (May 31):** Mostly sunny with a high near 80°F. Slight chance of showers and thunderstorms in the afternoon, 40% chance of precipitation. West wind 8 to 15 mph.
 
- **Saturday Night:** Mostly clear after a slight chance of evening showers and thunderstorms. Low around 56°F. West wind 2 to 12 mph.

- **Sunday (June 1):** Sunny with a high near 80°F. Slight chance of rain showers in the afternoon. Northwest wind 2 to 6 mph.

- **Sunday Night:** Partly cloudy with a low around 58°F. Slight chance of rain showers. 

- **Monday (June 2):** Sunny with a high near 82°F. Northwest wind about 5 mph.

- **Monday Night:** Clear with a low around 59°F.

- **Tuesday (June 3):** Sunny with a high near 88°F. Southwest wind about 3 mph.

- **Tuesday Night:** Mostly clear with a low around 64°F.

- **Wednesday (June 4):** Sunny with a high near 91°F. Southwest wind about 5 mph.

- **Wednesday Night:** Mostly clear with a low around 68°F.

- **Thursday (June 5):** Mostly sunny with a high near 92°F. Slight chance of precipitation later in the day.

- **Thursday Night:** Partly cloudy with a low around 71°F. Slight chance of showers and thunderstorms.

Structured Output from LangGraph Agent + MCP servers

The output from the previous example is in free‐text format, which can be fine for quick, human‐readable feedback but often falls short when you need to integrate results into downstream systems. In many scenarios — such as logging, analytics, or API‐driven pipelines — structured outputs are not only more reliable but also dramatically easier to validate, parse, and automate. By enforcing a schema (for example, returning a JSON object with clearly named fields), you eliminate ambiguity around data types and field semantics. Client applications can then extract values directly without resorting to brittle string-matching or regular expressions.

To get structured output from create_react_agent, you have several approaches depending on which framework you're using. Here are the two main methods:

1. Using Output Parsers

from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

class WeatherForecastOutput(BaseModel):
 """Weather forecast for weekdays in a given location."""

 Monday: Optional[str] = Field(description="Weather forecast for Monday")
 Tuesday: Optional[str] = Field(description="Weather forecast for Tuesday")
 Wednesday: Optional[str] = Field(description="Weather forecast for Wednesday")
 Thursday: Optional[str] = Field(description="Weather forecast for Thursday")
 Friday: Optional[str] = Field(description="Weather forecast for Friday")

parser = PydanticOutputParser(pydantic_object=WeatherForecastOutput)

async def main(query: str):
 async with MultiServerMCPClient(server_params) as client:
 tools = client.get_tools()
 agent = create_react_agent(model, tools)
 parser_instructions = parser.get_format_instructions()
 messages = [
 ("system", "You must respond with a valid JSON object. " + parser_instructions),
 ("human", f"Respond the question: {query}")
 ]
 formatted_messages = [HumanMessage(content=msg[1]) if msg[0] == "human" else SystemMessage(content=msg[1]) for msg in messages]
 response = await agent.ainvoke({"messages": formatted_messages})
 
 return response

if __name__ == "__main__":
 response=asyncio.run(main(query))
 print('------------')
 print(parser.parse(last_message.content))

2. Using Structured Output in OpenAI Function Calling

from langchain.schema import HumanMessage, SystemMessage
import json

async def main(query: str):
 async with MultiServerMCPClient(server_params) as client:
 tools = client.get_tools()
 agent = create_react_agent(model, tools)
 system_message = SystemMessage(content="""
 Always respond in this JSON format:
 {
 "Monday": "weather forecast for Monday",
 "Tuesday": "weather forecast for Tuesday",
 "Wednesday": "weather forecast for Wednesday",
 "Thursday": "weather forecast for Thursday",
 "Friday": "weather forecast for Friday"
 }
 """)
 human_message = HumanMessage(content=f"Respond the question: {query}")
 messages = [system_message, human_message]
 response = await agent.ainvoke({"messages": messages})
 return response

if __name__ == "__main__":
 response=asyncio.run(main(query))
 print('------------')
 last_message = response["messages"][-1]
 print(last_message.content)

Now, the response to the query “Forecast the weather in Raleigh, NC next week.” is returned in JSON format.

{
 "Monday": "Sunny, with a high near 81. North wind around 5 mph.",
 "Tuesday": "Sunny, with a high near 86. Slight chance of precipitation (1%).",
 "Wednesday": "Sunny, with a high near 88. Slight chance of precipitation (5%).",
 "Thursday": "Sunny, with a high near 90. Slight chance of precipitation (10%).",
 "Friday": "Mostly sunny, with a high near 91. Chance of showers and thunderstorms after 2 pm (30%)."
}

You can access any day’s forecast simply by referencing its key. For example, in Python:

weather_forecast = json.loads(last_message.content)
print(weather_forecast['Monday'])

Join thousands of data leaders on the AI newsletter. Join over 80,000 subscribers and keep up to date with the latest developments in AI. From research to projects and ideas. If you are building an AI startup, an AI-related product, or a service, we invite you to consider becoming a sponsor.

Published via Towards AI