DEV Community: Kasi Yaswanth

Synchronizing Knowledge Graphs

Kasi Yaswanth — Mon, 17 Aug 2026 03:41:14 +0000

Imagine you're building a support bot that uses a LangGraph agent to guide users through troubleshooting their smart home devices. The bot works flawlessly when users interact with it sequentially, but things start to fall apart when multiple users engage with it concurrently. The agent begins to forget the context of previous conversations, leading to confusing and incorrect responses. Upon further investigation, you realize that the issue stems from the fact that each conversation thread is updating the agent's knowledge graph independently, resulting in temporal inconsistencies.

To mitigate this problem, you can leverage MCP's Resource primitive to synchronize the knowledge graphs across concurrent conversation threads and tool invocations. The Resource primitive allows you to define a shared knowledge graph that can be accessed and updated by multiple agents, ensuring that all agents have a consistent view of the world.

Let's dive into the details of how you can use the Resource primitive to synchronize knowledge graphs. First, you need to define a Resource that represents the shared knowledge graph. You can do this by creating an instance of the Resource class and specifying the type of resource you want to create. In this case, you want to create a knowledge graph resource, so you would use the KnowledgeGraph type.

from mcp import Resource

# Define a knowledge graph resource
knowledge_graph_resource = Resource(
    type="KnowledgeGraph",
    name="smart_home_devices"
)

Next, you need to create a LangGraph agent that uses the knowledge graph resource. You can do this by creating an instance of the LangGraph class and specifying the knowledge graph resource as the agent's knowledge graph.

from langgraph import LangGraph

# Create a LangGraph agent that uses the knowledge graph resource
agent = LangGraph(
    knowledge_graph=knowledge_graph_resource
)

Now, when multiple conversation threads or tool invocations update the agent's knowledge graph, they will all be updating the same shared knowledge graph resource. This ensures that all agents have a consistent view of the world, mitigating the temporal inconsistencies that were causing the support bot to forget context.

To take it a step further, you can use the add_node and add_conditional_edges methods to define the structure of the knowledge graph and specify how the agent should reason about the devices. For example, you could add nodes for each device type and conditional edges that specify how the agent should troubleshoot each device.

# Add nodes for each device type
agent.knowledge_graph.add_node("device_type", "router")
agent.knowledge_graph.add_node("device_type", "modem")
agent.knowledge_graph.add_node("device_type", "smart_tv")

# Add conditional edges that specify how to troubleshoot each device
agent.knowledge_graph.add_conditional_edges(
    "device_type",
    "router",
    [
        ("is_powered_on", "check_power_cord"),
        ("is_connected_to_internet", "check_wifi_signal")
    ]
)

By using the Resource primitive to synchronize knowledge graphs across concurrent conversation threads and tool invocations, you can ensure that your support bot provides accurate and consistent responses, even in the face of multiple concurrent interactions.

One practical gotcha to watch out for when using the Resource primitive is that it can introduce additional latency into your system, since all agents need to access and update the shared knowledge graph resource. To mitigate this, you can use techniques like caching or batching to reduce the number of requests to the knowledge graph resource.

As we continue to explore the complexities of building agentic AI systems, we'll be diving deeper into the challenges of scaling and deploying these systems in real-world environments. Tomorrow, we'll be looking at how to use MCP's tooling to streamline the development and testing of LangGraph agents, and how to integrate these agents with other AI systems to create more sophisticated and autonomous applications.

LangGraph API Failure

Kasi Yaswanth — Sun, 16 Aug 2026 03:39:39 +0000

We've all been there - you've built an agentic AI system using LangGraph and MCP, and it's been humming along just fine, until one day, a third-party API that one of your MCP Tool primitives relies on suddenly changes its response format. Your agent, which was previously adapting beautifully to user input, is now stuck in an infinite loop, unable to recover from the unexpected API response. This is exactly what happened to us recently, and in this post, we'll dive into the postmortem analysis of the failure, and explore strategies for implementing robustness against similar failures in the future.

The specific issue we encountered was with an MCP Tool primitive that used the Google Maps API to get the current traffic conditions. The primitive was designed to adapt to changes in traffic conditions by adjusting the recommended route accordingly. However, when the Google Maps API suddenly changed its response format, our primitive was unable to parse the response, and our LangGraph agent was unable to adapt to the new format. The agent kept trying to execute the primitive, but the primitive kept failing, causing the agent to loop indefinitely.

To understand why this happened, let's take a look at the code for the MCP Tool primitive:

import json
from mcp import Tool

class TrafficConditions(Tool):
    def __init__(self, api_key):
        self.api_key = api_key

    def execute(self, context):
        url = f"https://maps.googleapis.com/maps/api/traffic/json?key={self.api_key}"
        response = requests.get(url)
        data = json.loads(response.content)
        traffic_conditions = data["traffic"]["conditions"]
        context["traffic_conditions"] = traffic_conditions
        return context

The issue here is that the primitive assumes a specific response format from the Google Maps API, and when that format changes, the primitive fails. To make the primitive more robust, we can add some error handling to handle unexpected response formats. We can also use the LangGraph add_conditional_edges method to specify alternative edges to take when the primitive fails.

Here's an updated version of the primitive that includes error handling and alternative edges:

import json
from mcp import Tool
from langgraph import StateGraph

class TrafficConditions(Tool):
    def __init__(self, api_key):
        self.api_key = api_key

    def execute(self, context):
        try:
            url = f"https://maps.googleapis.com/maps/api/traffic/json?key={self.api_key}"
            response = requests.get(url)
            data = json.loads(response.content)
            traffic_conditions = data["traffic"]["conditions"]
            context["traffic_conditions"] = traffic_conditions
            return context
        except KeyError:
            # handle unexpected response format
            context["error"] = "Unexpected response format from Google Maps API"
            return context

# create a LangGraph state graph
graph = StateGraph()

# add a node for the TrafficConditions primitive
node = graph.add_node("TrafficConditions", TrafficConditions(api_key="YOUR_API_KEY"))

# add a conditional edge to handle the case where the primitive fails
graph.add_conditional_edges(node, [
    ("error", "Unexpected response format from Google Maps API", "ErrorHandlingNode")
])

In this updated version, we've added a try-except block to handle the case where the Google Maps API returns an unexpected response format. We've also added a conditional edge to specify an alternative edge to take when the primitive fails. This allows the LangGraph agent to adapt to the failure and take a different path.

One practical gotcha to watch out for when implementing robustness against API changes is to make sure to test your primitives and agents thoroughly after making changes. It's easy to introduce new bugs or unexpected behavior when adding error handling or alternative edges, so make sure to test your system thoroughly to ensure that it's working as expected.

As we continue to build and deploy agentic AI systems, we'll encounter more and more complex challenges and failures. Tomorrow, we'll explore another critical aspect of building robust agentic AI systems, and dive deeper into the complexities of designing and implementing adaptive agents that can thrive in a rapidly changing world.

Optimizing Latency with MCP Caching

Kasi Yaswanth — Sat, 15 Aug 2026 03:31:57 +0000

I recently spent a frustrating afternoon debugging a support bot built on top of LangGraph, where the bot would occasionally take an inordinate amount of time to respond to user queries. The bot's workflow involved multiple tool calls to external services, such as entity disambiguation and sentiment analysis, which were implemented using the Model Context Protocol (MCP). After digging into the logs, I noticed that the bot was spending a significant amount of time waiting for these tool calls to complete, even when the input data was identical to previous requests. This was causing the bot's overall latency to spike, leading to a poor user experience.

The root cause of the issue was that the bot was not leveraging MCP's caching mechanisms for frequent tool calls. Every time the bot needed to make a tool call, it would create a new request and wait for the response, even if the same request had been made before. This was not only inefficient but also unnecessary, as the results of these tool calls were often deterministic and didn't change between requests.

To address this issue, I started by identifying the most frequently called tools in the bot's workflow and configuring MCP to cache their results. This involved adding a cache layer on top of the tool calls, which would store the results of previous requests and return them immediately if the same request was made again. I used the MCP_CACHE environment variable to enable caching for the specific tools that needed it.

Here's an example of how I implemented caching for a sentiment analysis tool using MCP's caching mechanisms:

import os
import langgraph as lg
from langgraph.tools import MCPTool

# Enable caching for the sentiment analysis tool
os.environ['MCP_CACHE'] = 'sentiment_analysis_tool'

# Define the sentiment analysis tool
class SentimentAnalysisTool(MCPTool):
    def __init__(self):
        super().__init__('sentiment_analysis_tool')

    def run(self, input_text):
        # Make the tool call and cache the result
        cached_result = self.cache.get(input_text)
        if cached_result is not None:
            return cached_result

        # If the result is not cached, make the tool call and store the result
        result = self.make_tool_call(input_text)
        self.cache.set(input_text, result)
        return result

# Create a LangGraph workflow that uses the sentiment analysis tool
workflow = lg.StateGraph()
sentiment_tool = SentimentAnalysisTool()

# Add a node to the workflow that uses the sentiment analysis tool
workflow.add_node('sentiment_analysis', sentiment_tool.run)

# Add conditional edges to the workflow based on the sentiment analysis result
workflow.add_conditional_edges('sentiment_analysis', [
    (lambda x: x > 0.5, 'positive_sentiment'),
    (lambda x: x < -0.5, 'negative_sentiment')
])

# Run the workflow and print the result
result = workflow.run('This is a great product!')
print(result)

By leveraging MCP's caching mechanisms, I was able to significantly reduce the latency of the support bot and improve the overall user experience. One practical gotcha to watch out for when implementing caching is to ensure that the cache is properly invalidated when the underlying data changes. In this case, I made sure to invalidate the cache whenever the sentiment analysis model was updated or retrained.

As we continue to build more complex agentic systems, optimizing latency and performance will become increasingly important. Tomorrow, we'll explore another critical aspect of building scalable agentic systems, and how to apply the lessons learned from this experience to tackle even more challenging problems.

Cascading Failures in Multi-Agent Systems

Kasi Yaswanth — Fri, 14 Aug 2026 04:42:06 +0000

I still remember the day our support bot, powered by a LangGraph agent workflow, started behaving erratically. Customers would ask for help with their orders, and the bot would respond with irrelevant information or, worse, loop indefinitely. After digging into the logs, we discovered that the issue was caused by a cascading failure in our multi-agent system. One of the downstream services, responsible for fetching order data, had gone offline, and our agent workflow didn't know how to handle it.

As we investigated further, we realized that our agents were tightly coupled to the availability of these downstream services. If one service failed, the entire workflow would come to a grinding halt. We needed a way to detect when a service was down and recover from it gracefully. That's when we turned to MCP's Tool primitive.

The Tool primitive in MCP allows us to define a reusable piece of functionality that can be used across multiple agents. In our case, we created a Tool that would check the status of our downstream services and notify the agent workflow if any of them were unavailable. We could then use this information to recover from the failure and provide a better experience for our customers.

Here's an example of how we implemented this using Python and the MCP API:

import langgraph as lg
from mcp import Tool, ToolRequest

# Define the Tool that checks the status of our downstream services
class ServiceChecker(Tool):
    def __init__(self, service_urls):
        self.service_urls = service_urls

    def check_services(self):
        available_services = []
        for url in self.service_urls:
            try:
                # Simulate a request to the service
                response = requests.head(url)
                if response.status_code == 200:
                    available_services.append(url)
            except requests.RequestException:
                pass
        return available_services

# Create a LangGraph agent workflow that uses the ServiceChecker Tool
def create_workflow():
    graph = lg.StateGraph()
    service_checker = ServiceChecker(["https://service1.example.com", "https://service2.example.com"])

    # Add a node to the graph that checks the status of the services
    graph.add_node("check_services", service_checker.check_services)

    # Add a conditional edge to the graph that depends on the result of the service check
    graph.add_conditional_edges("check_services", [
        (lambda result: len(result) == 2, "proceed_with_workflow"),
        (lambda result: len(result) < 2, "handle_service_outage")
    ])

    # Add a node to the graph that handles the service outage
    graph.add_node("handle_service_outage", lambda: "Service outage detected. Please try again later.")

    return graph

# Create the workflow and execute it
workflow = create_workflow()
result = workflow.execute()
print(result)

In this example, we define a ServiceChecker Tool that takes a list of service URLs and checks their status. We then create a LangGraph agent workflow that uses this Tool to check the status of the services and recover from any outages.

One practical gotcha we learned from this experience is that it's essential to consider the latency of the service checks when designing the agent workflow. If the service checks take too long, the workflow may timeout or become unresponsive. To mitigate this, we can use techniques like caching or asynchronous service checks to reduce the latency.

As we continue to build and deploy more complex multi-agent systems, we'll need to develop strategies for handling failures and exceptions. Tomorrow, we'll explore another critical aspect of building robust agentic AI systems, and I'm excited to share our learnings and insights with you.

LangGraph Episodic Memory

Kasi Yaswanth — Thu, 13 Aug 2026 04:44:33 +0000

I recently encountered a frustrating issue with a support bot I'd built using LangGraph. The bot was designed to help users troubleshoot common problems with their smart home devices. However, I noticed that it would often forget the context of the conversation, asking the user to repeat information they'd already provided. This wasn't just annoying - it was also a major obstacle to providing effective support. The bot's inability to retain context-dependent knowledge across multiple interactions made it seem like it was starting from scratch every time the user asked a follow-up question.

After digging into the code, I realized that the problem lay in the way I was handling the bot's episodic memory. Episodic memory refers to the ability of an agent to recall specific events or experiences from the past. In the case of my support bot, this meant remembering the details of the user's previous interactions, such as the device they were trying to troubleshoot and the steps they'd already taken. I was using a simple in-memory store to keep track of this information, which was being lost whenever the conversation ended or the bot was restarted.

To fix this issue, I decided to use the Resource primitive from the Model Context Protocol (MCP) to persist and retrieve the bot's episodic memory. The Resource primitive allows you to define a shared context that can be accessed and updated by multiple components within an agent. This made it ideal for storing the bot's episodic memory, as it would allow me to persist the memory across multiple conversations and user interactions.

Here's an example of how I implemented the bot's episodic memory using the Resource primitive:

import langgraph as lg
from mcp import Resource

# Define a resource to store the bot's episodic memory
memory_resource = Resource("episodic_memory")

# Define a function to update the episodic memory
def update_memory(user_id, conversation_history):
    memory_resource.update(user_id, conversation_history)

# Define a function to retrieve the episodic memory
def get_memory(user_id):
    return memory_resource.get(user_id)

# Create a LangGraph agent that uses the episodic memory
class SupportBot(lg.Agent):
    def __init__(self):
        super().__init__()
        self.memory_resource = memory_resource

    def respond(self, user_id, input_text):
        # Retrieve the user's conversation history from the episodic memory
        conversation_history = get_memory(user_id)

        # Update the conversation history with the new input
        if conversation_history is None:
            conversation_history = []
        conversation_history.append(input_text)
        update_memory(user_id, conversation_history)

        # Use the conversation history to inform the bot's response
        if "troubleshoot" in input_text:
            # Retrieve the device type from the conversation history
            device_type = None
            for message in conversation_history:
                if "device type" in message:
                    device_type = message.split(" ")[-1]
                    break

            # Provide troubleshooting steps based on the device type
            if device_type == "thermostat":
                return "Try restarting the thermostat and checking the temperature settings."
            elif device_type == "light bulb":
                return "Try replacing the light bulb and checking the power source."

        # If the bot doesn't understand the input, ask for clarification
        return "I didn't understand that. Can you please rephrase?"

# Create an instance of the support bot and test it
bot = SupportBot()
print(bot.respond("user1", "I'm having trouble with my thermostat"))
print(bot.respond("user1", "The device type is thermostat"))
print(bot.respond("user1", "troubleshoot"))

In this example, the SupportBot class uses the Resource primitive to store and retrieve the user's conversation history. The update_memory function updates the conversation history with new input, and the get_memory function retrieves the conversation history for a given user. The bot uses this conversation history to inform its responses, providing more effective support by retaining context-dependent knowledge across multiple interactions.

One practical gotcha to watch out for when implementing episodic memory using the Resource primitive is that it can be easy to over-engineer the memory store. It's tempting to try to store every possible piece of information that might be relevant to the conversation, but this can lead to a bloated and inefficient memory store. Instead, focus on storing only the most important information, and use clever indexing and retrieval strategies to make it easy to access and update the memory.

As we continue to explore the possibilities of agentic AI, we'll be looking at even more advanced techniques for managing context and memory. Tomorrow, we'll dive into the challenges of integrating multiple knowledge sources and reasoning systems to create a truly robust and flexible agent.

Evaluating Agent Effectiveness

Kasi Yaswanth — Wed, 12 Aug 2026 04:41:05 +0000

I was working on a support bot that used a LangGraph agent to troubleshoot customer issues. The bot was supposed to guide the customer through a series of questions to identify the root cause of their problem, but I noticed that it was often getting stuck in an infinite loop, asking the same questions over and over. The issue wasn't with the graph itself, but with the way the agent was evaluating its progress towards the conversation goal. It was focusing too much on the short-term objective of answering the current question, rather than the long-term goal of resolving the customer's issue.

This is a classic problem in agentic AI, where the agent gets bogged down in the details of the current task and loses sight of the bigger picture. To solve this, I needed to find a way to evaluate the agent's effectiveness through indirect objectives, rather than just its immediate actions. That's where MCP's Resource primitive came in.

The Resource primitive in MCP allows you to define abstract resources that the agent can use to track its progress towards a goal. These resources can be anything from a simple counter to a complex data structure, and they can be used to evaluate the agent's performance over time. In my case, I defined a resource called "CustomerProgress" that tracked the customer's progress through the troubleshooting process.

Here's an example of how I used the Resource primitive in Python:

import langgraph as lg
from mcp import Resource, add_resource

# Define the CustomerProgress resource
class CustomerProgress(Resource):
    def __init__(self):
        self.progress = 0

    def update(self, new_progress):
        self.progress = new_progress

# Create a LangGraph agent
agent = lg.Agent()

# Add the CustomerProgress resource to the agent
add_resource(agent, CustomerProgress())

# Define a state graph for the troubleshooting process
graph = lg.StateGraph()
graph.add_node("start", "Welcome to the troubleshooting process!")
graph.add_node("question1", "What is your issue?")
graph.add_node("question2", "Can you provide more details?")
graph.add_node("resolution", "I think I have found the solution to your issue.")

# Add conditional edges to the graph based on the customer's responses
graph.add_conditional_edges("question1", "question2", lambda x: x == "yes")
graph.add_conditional_edges("question2", "resolution", lambda x: x == "yes")

# Define a checkpointer to update the CustomerProgress resource
def checkpoint_progress(state):
    if state == "question1":
        agent.resources["CustomerProgress"].update(0.2)
    elif state == "question2":
        agent.resources["CustomerProgress"].update(0.5)
    elif state == "resolution":
        agent.resources["CustomerProgress"].update(1.0)

# Add the checkpointer to the graph
graph.add_checkpoint(checkpoint_progress)

# Run the agent
agent.run(graph)

In this example, the CustomerProgress resource is used to track the customer's progress through the troubleshooting process. The checkpoint_progress function updates the resource based on the current state of the graph, and the add_checkpoint method adds this function to the graph as a checkpoint. The agent can then use this resource to evaluate its progress towards the conversation goal.

One practical gotcha I learned while working with the Resource primitive is that it's easy to get caught up in defining complex resources and forget to actually use them to evaluate the agent's performance. Make sure to define clear objectives and metrics for your resources, and use them to inform your agent's decision-making process.

As I continue to work on my support bot, I'm excited to explore more ways to use MCP's resources and LangGraph's state graphs to create a more effective and efficient conversation flow. Tomorrow, I'll be diving deeper into the world of agentic AI and exploring new ways to optimize my agent's performance.

Scalable MCP Server

Kasi Yaswanth — Tue, 11 Aug 2026 04:11:46 +0000

I still remember the day our LangGraph agent deployment came crashing down due to a simple yet devastating issue - our MCP server couldn't handle the sheer volume of tool registrations and endpoint requests. We had built a robust agentic AI system, but our server architecture was a bottleneck, causing agents to fail or behave erratically. The specific error message that haunted us was "Max tool registrations exceeded," which seemed straightforward but led to a complex problem-solving journey.

As we delved deeper, we realized that our MCP server was not designed to scale with our growing LangGraph agent deployment. We had a fixed set of tools registered, and each time a new agent was spun up, it would attempt to register its own tools, leading to a rapid exhaustion of available slots. Moreover, our endpoint routing was static, which meant that agents would often try to access tools that were already in use or not available, resulting in further failures.

To tackle this issue, we set out to design a highly available MCP server architecture that could automate tool registration and dynamically route endpoint requests. We began by introducing a load balancer to distribute incoming requests across multiple MCP server instances. This ensured that no single server was overwhelmed, and we could easily add or remove instances as needed.

Next, we developed an automated tool registration system using the MCP API. We created a registry service that would listen for new tool registrations and dynamically update the available tools on each MCP server instance. This allowed us to scale our toolset without manual intervention and ensured that agents could always find the tools they needed.

Here's an example of how we implemented the automated tool registration using Python and the MCP API:

import MCP

# Create an MCP client instance
mcp_client = MCP.Client()

# Define a function to register a new tool
def register_tool(tool_name, tool_description):
    # Create a new tool registration request
    request = MCP.ToolRegistrationRequest(
        name=tool_name,
        description=tool_description
    )

    # Send the registration request to the MCP server
    response = mcp_client.register_tool(request)

    # Check if the registration was successful
    if response.status == MCP.ToolRegistrationStatus.SUCCESS:
        print(f"Tool {tool_name} registered successfully")
    else:
        print(f"Failed to register tool {tool_name}")

# Define a function to update the available tools on an MCP server instance
def update_available_tools(mcp_server_instance):
    # Get the current list of available tools
    available_tools = mcp_client.get_available_tools(mcp_server_instance)

    # Get the list of newly registered tools
    new_tools = mcp_client.get_newly_registered_tools()

    # Update the available tools on the MCP server instance
    mcp_client.update_available_tools(mcp_server_instance, available_tools + new_tools)

# Create a registry service that listens for new tool registrations
registry_service = MCP.RegistryService()

# Define a callback function to handle new tool registrations
def on_new_tool_registration(tool_name, tool_description):
    # Register the new tool
    register_tool(tool_name, tool_description)

    # Update the available tools on all MCP server instances
    for mcp_server_instance in mcp_client.get_mcp_server_instances():
        update_available_tools(mcp_server_instance)

# Start the registry service
registry_service.start(on_new_tool_registration)

With our automated tool registration system in place, we turned our attention to dynamic endpoint routing. We introduced a routing service that would inspect incoming requests and direct them to the most suitable MCP server instance based on factors like tool availability and server load.

One practical gotcha we encountered during this process was the need to implement idempotent tool registration. Since our registry service would retry failed registrations, we had to ensure that registering a tool multiple times would not lead to duplicate entries or other inconsistencies. We achieved this by using a unique identifier for each tool and checking for existing registrations before attempting to register a new tool.

As we look to the future, our highly available MCP server architecture will provide a solid foundation for further innovations in LangGraph agent development, enabling us to push the boundaries of what's possible with agentic AI and explore new applications and use cases that we can't yet imagine. Tomorrow, we'll dive deeper into the intricacies of LangGraph agent development and explore new ways to harness the power of MCP and LangGraph to build more sophisticated and capable AI systems.

MCP Resource for LangGraph

Kasi Yaswanth — Mon, 10 Aug 2026 04:22:08 +0000

I was recently debugging a support bot built with LangGraph, where the agent would consistently forget the context of the conversation after a few turns. The bot would ask the user to repeat information they had already provided, leading to a frustrating experience. Upon closer inspection, I realized that the issue stemmed from the fact that the agent's knowledge graph was not being properly updated and retrieved across multiple tool calls. The agent was essentially starting from scratch every time it received a new input, which made it impossible to maintain a coherent conversation.

This is where MCP's Resource primitive comes into play. The Resource primitive allows us to define a knowledge graph that can be efficiently updated and retrieved across multiple tool calls and conversation turns. By using the Resource primitive, we can create a centralized store of information that the agent can draw upon to inform its decisions and actions.

To illustrate this, let's consider an example where we're building a support bot that helps users troubleshoot issues with their devices. We can define a Resource that represents the user's device information, which includes details such as the device type, operating system, and any error messages they've encountered. We can then use this Resource to inform the agent's responses and actions, such as providing troubleshooting steps or escalating the issue to a human support agent.

Here's an example of how we might implement this using MCP's Resource primitive and LangGraph:

import langgraph
from mcp import Resource

# Define the device information Resource
device_info_resource = Resource(
    name="device_info",
    attributes=[
        {"name": "device_type", "type": "string"},
        {"name": "operating_system", "type": "string"},
        {"name": "error_messages", "type": "list<string>"}
    ]
)

# Create a LangGraph agent that uses the device information Resource
agent = langgraph.Agent()

# Define a node in the agent's graph that retrieves the device information Resource
device_info_node = agent.add_node(
    name="get_device_info",
    func=lambda: device_info_resource.get()
)

# Define a node in the agent's graph that updates the device information Resource
update_device_info_node = agent.add_node(
    name="update_device_info",
    func=lambda device_type, operating_system, error_messages: device_info_resource.update(
        device_type=device_type,
        operating_system=operating_system,
        error_messages=error_messages
    )
)

# Define a conditional edge between the nodes
agent.add_conditional_edges(
    from_node=device_info_node,
    to_node=update_device_info_node,
    condition=lambda device_info: device_info is None
)

# Run the agent
agent.run()

In this example, we define a Resource called device_info that represents the user's device information. We then create a LangGraph agent that uses this Resource to inform its decisions and actions. The agent has two nodes: one that retrieves the device information Resource, and another that updates the Resource. The agent uses a conditional edge to determine which node to execute based on the current state of the Resource.

One practical gotcha to watch out for when using MCP's Resource primitive is that it can be easy to over-engineer the Resource, leading to a complex and brittle knowledge graph. It's essential to strike a balance between providing enough information to inform the agent's decisions and actions, and avoiding unnecessary complexity.

As we continue to build more sophisticated agentic AI systems, we'll need to explore new ways to manage complexity and ensure that our agents are able to effectively retrieve and update information across multiple tool calls and conversation turns. Tomorrow, we'll dive deeper into the challenges of scaling LangGraph agents to handle increasingly complex tasks and scenarios.

Day 30/30: Building End-to-End Support Ticket Triage

Kasi Yaswanth — Sun, 09 Aug 2026 04:10:37 +0000

I still remember the day our support-ticket triage agent went live, only to start forgetting the context of conversations after a few exchanges. Customers would report issues, and our agent would respond with generic solutions, completely unaware of the previous messages. It was as if the agent had no memory. We quickly realized that our implementation was missing a crucial piece: the ability to maintain state across multiple interactions.

To fix this, we turned to LangGraph's StateGraph and MCP's tools for building agentic AI systems. We designed a state machine that could transition between different stages of a support ticket, from initial report to resolution. Each stage would have its own set of possible actions and next steps, depending on the customer's input. We used LangGraph's add_node and add_conditional_edges methods to define these states and transitions.

Here's an example of how we defined the initial state and its transitions:

import langgraph as lg
from mcp import tools

# Create a new StateGraph
graph = lg.StateGraph()

# Define the initial state
initial_state = graph.add_node("initial_report")
initial_state.set_description("Customer reports an issue")

# Define possible next states based on customer input
next_states = {
    "bug_report": graph.add_node("bug_report"),
    "feature_request": graph.add_node("feature_request"),
    "general_inquiry": graph.add_node("general_inquiry")
}

# Add conditional edges between states
for intent, next_state in next_states.items():
    graph.add_conditional_edges(initial_state, next_state, condition=f"intent == '{intent}'")

# Define the MCP tools and resources needed for each state
tools.register_tool("bug_report", tools.BugReportTool())
tools.register_tool("feature_request", tools.FeatureRequestTool())
tools.register_resource("knowledge_base", tools.KnowledgeBaseResource())

# Use MCP's checkpointers to save the current state and context
checkpointer = tools.Checkpointer("support_ticket_triage")

With this implementation, our support-ticket triage agent could now maintain context across multiple interactions and respond accordingly. However, we soon encountered another issue: the agent would sometimes get stuck in an infinite loop, repeatedly asking the customer for more information without making progress towards resolving the issue.

We learned that this was due to the agent's lack of a clear goal or objective. To fix this, we had to define a set of success criteria for each state and ensure that the agent was working towards achieving those goals. This involved adding more conditional edges and next states to the StateGraph, as well as using MCP's tools and resources to provide the agent with more context and information.

One practical gotcha we encountered was the need to carefully balance the agent's use of memory and context. If the agent retained too much information, it would become slow and unresponsive. On the other hand, if it forgot too much, it would lose context and struggle to respond effectively. This balance is crucial in building effective agentic AI systems, and it's a lesson we'll carry forward as we continue to explore the possibilities of LangGraph and MCP.

As we look to the future, we're excited to see how these technologies will continue to evolve and improve, enabling us to build even more sophisticated and effective AI systems that can tackle complex challenges and make a real impact on people's lives. Tomorrow, we'll be exploring new frontiers in agentic AI, and we can't wait to see what the future holds.

Day 29/30: Deploying LangGraph + MCP

Kasi Yaswanth — Sat, 08 Aug 2026 04:05:10 +0000

So you've finally gotten your LangGraph + MCP agent working on your local machine - it's responding to queries, adapting to new information, and generally behaving itself. But as soon as you try to deploy it to a server, things start to fall apart. The agent crashes after a few hours, or it starts producing bizarre, out-of-context responses. You're left scratching your head, wondering why your carefully crafted agent can't seem to stay up and running in production.

The problem often lies in the way we've structured our agent's lifecycle. When we're running locally, it's easy to just spin up the agent, test it out, and then shut it down when we're done. But in a production environment, our agent needs to be able to recover from failures, adapt to changing circumstances, and handle a constant stream of incoming requests. This requires a more robust approach to deployment, one that takes into account the realities of distributed systems and the unpredictability of the real world.

One key concept here is the idea of a "service" - a long-running process that provides some kind of functionality to the outside world. In our case, we want to deploy our LangGraph + MCP agent as a service that can receive incoming requests, process them, and respond accordingly. To do this, we'll need to use a combination of tools and techniques to ensure our agent stays up and running, even in the face of unexpected failures or changes.

Here's an example of how we might structure our agent as a service, using Python and the LangGraph + MCP APIs:

import langgraph
from mcp import ModelContextProtocol
import os
import signal
import sys

# Set up our LangGraph + MCP agent
agent = langgraph.Agent()
mcp = ModelContextProtocol()

# Define a function to handle incoming requests
def handle_request(request):
    # Use our agent to process the request and generate a response
    response = agent.process_request(request)
    return response

# Define a function to run our agent as a service
def run_service():
    # Set up a signal handler to catch SIGTERM and SIGINT
    def signal_handler(sig, frame):
        print("Received signal, shutting down...")
        agent.shutdown()
        sys.exit(0)

    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)

    # Start our agent and begin listening for incoming requests
    agent.start()
    while True:
        request = mcp.receive_request()
        response = handle_request(request)
        mcp.send_response(response)

# Run our agent as a service
if __name__ == "__main__":
    run_service()

In this example, we define a run_service function that sets up our agent and begins listening for incoming requests. We use the signal module to catch SIGTERM and SIGINT signals, which allows us to cleanly shut down our agent when the service is terminated. We also define a handle_request function that uses our agent to process incoming requests and generate responses.

One practical gotcha to watch out for when deploying a LangGraph + MCP agent as a service is the need to properly handle checkpointing and recovery. If our agent crashes or is terminated unexpectedly, we'll need to be able to recover its state and resume processing from where we left off. This can be achieved using the langgraph.Checkpointer class, which allows us to save and restore our agent's state at regular intervals.

As we look to the future, we'll be exploring even more advanced techniques for deploying and managing LangGraph + MCP agents in production environments. With the ability to build robust, scalable agents that can adapt to changing circumstances, we'll be able to tackle even more complex and challenging problems in areas like natural language processing, computer vision, and beyond. Tomorrow, we'll be taking a closer look at how to integrate our agents with other systems and services, and how to use techniques like reinforcement learning to optimize their performance.

Day 28/30: Debugging Agent Failures

Kasi Yaswanth — Fri, 07 Aug 2026 04:47:57 +0000

It's 2am and your support bot, powered by LangGraph and MCP, has suddenly stopped responding to user queries. The error logs are filled with generic messages about "context not found" and "conditional edge failures," but nothing seems to point to the root cause of the issue. You've tried restarting the service, checking the database connections, and even re-running the last successful conversation, but to no avail. The problem persists, and you're no closer to understanding what's going wrong.

As you dig deeper, you realize that the issue lies in the way your agent is handling context. Specifically, it seems to be forgetting the conversation history, causing it to loop back to the initial state and lose track of the user's intent. But why is this happening? Is it a problem with the StateGraph implementation, the add_conditional_edges method, or something else entirely?

To get to the bottom of this, you need to add more observability to your agent. This means logging specific events and data points that can help you understand what's happening during the conversation flow. In this case, you decide to log the following:

The current state of the conversation (e.g., initial, context_set, response_generated)
The input prompt or user query
The output response generated by the agent
Any errors or exceptions that occur during processing
The context variables and their values at each step

By logging these events, you can reconstruct the conversation flow and identify where things are going wrong. Here's an example of how you might implement this using LangGraph and MCP:

import langgraph as lg
from mcp import tools

# Define the conversation states
states = ["initial", "context_set", "response_generated"]

# Create a StateGraph instance
graph = lg.StateGraph()

# Add nodes for each state
for state in states:
    graph.add_node(state)

# Add conditional edges between states
graph.add_conditional_edges([
    ("initial", "context_set", lambda ctx: ctx.get("user_query") is not None),
    ("context_set", "response_generated", lambda ctx: ctx.get("context_variables") is not None)
])

# Define a logging function to track conversation events
def log_event(event_type, data):
    print(f"{event_type}: {data}")

# Define the conversation flow
def conversation_flow(ctx):
    log_event("state_transition", ctx.get("current_state"))
    if ctx.get("current_state") == "initial":
        log_event("input_prompt", ctx.get("user_query"))
        # Set context variables and transition to context_set state
        ctx["context_variables"] = {"user_query": ctx["user_query"]}
        ctx["current_state"] = "context_set"
    elif ctx.get("current_state") == "context_set":
        log_event("context_variables", ctx.get("context_variables"))
        # Generate response and transition to response_generated state
        response = tools.generate_response(ctx["context_variables"])
        ctx["response"] = response
        ctx["current_state"] = "response_generated"
    elif ctx.get("current_state") == "response_generated":
        log_event("output_response", ctx.get("response"))
        # Return the final response
        return ctx["response"]

# Run the conversation flow
ctx = {"current_state": "initial", "user_query": "What is the weather like today?"}
response = conversation_flow(ctx)

# Check the logs to see what happened during the conversation

With this logging in place, you can now see exactly what's happening during the conversation flow. You notice that the context_variables are not being set correctly, causing the agent to lose track of the conversation history. You can fix this by modifying the conversation_flow function to properly set the context variables.

One practical gotcha to watch out for is that logging too much data can be overwhelming and make it harder to debug issues. It's essential to strike a balance between logging enough information to be useful and not so much that it becomes noise. In this case, logging the conversation states, input prompts, output responses, and context variables provides a good balance of information without being too verbose.

As you continue to build and refine your agentic AI systems, you'll encounter more complex challenges that require innovative solutions. Tomorrow, we'll explore another critical aspect of building reliable and efficient agents, one that will help you take your systems to the next level.

Day 27/30: Securing MCP Servers

Kasi Yaswanth — Thu, 06 Aug 2026 05:39:25 +0000

I still remember the day our support bot, powered by LangGraph and MCP, started behaving erratically. It would suddenly switch topics or reveal sensitive information to the wrong users. After digging through the logs, we discovered the issue: our MCP server was not properly secured, allowing unauthorized access to our language model's capabilities. The root cause was our over-reliance on the language model itself to manage consent and credentials. We had assumed that the LLM's understanding of context and user intent would suffice as a security model. It didn't.

The problem arose from the fact that our MCP server was using a single, unrestricted credential to interact with the language model. This meant that any user could potentially access any piece of information or perform any action, as long as they could craft a convincing prompt. To fix this, we needed to implement proper consent flows and scoped credentials.

Consent flows are essential in ensuring that users are aware of what data is being collected and how it will be used. In the context of an MCP server, this means defining clear boundaries for what actions can be performed by which users. Scoped credentials take this a step further by restricting access to specific parts of the system or data based on the user's role or identity.

For example, in our support bot, we wanted to ensure that only authorized personnel could access sensitive customer information. We achieved this by implementing a consent flow that required users to explicitly grant permission before such information could be accessed. We then used scoped credentials to restrict access to this information, so that even if an unauthorized user managed to bypass the consent flow, they would still be denied access due to the lack of appropriate credentials.

Here's a simplified example of how we might implement this in Python, using the MCP API to define a consent flow and scoped credentials:

import MCP

# Define a consent flow for accessing sensitive customer information
def consent_flow(user_id, action):
    # Check if the user has granted consent for the specified action
    if MCP.has_consent(user_id, action):
        return True
    else:
        # Prompt the user for consent if it hasn't been granted
        MCP.prompt_consent(user_id, action)
        return MCP.has_consent(user_id, action)

# Define scoped credentials for accessing sensitive customer information
def scoped_credentials(user_id):
    # Check the user's role and grant access accordingly
    if MCP.get_user_role(user_id) == "support_agent":
        return MCP.grant_access("customer_info")
    else:
        return MCP.deny_access("customer_info")

# Example usage
user_id = "example_user"
action = "access_customer_info"

if consent_flow(user_id, action) and scoped_credentials(user_id):
    # Access the sensitive customer information
    print(MCP.access_resource("customer_info"))
else:
    print("Access denied")

In this example, the consent_flow function checks if the user has granted consent for the specified action, and prompts them for consent if necessary. The scoped_credentials function grants or denies access to the sensitive customer information based on the user's role.

One practical gotcha we learned from this experience is the importance of regularly reviewing and updating our consent flows and scoped credentials. As our system evolves and new features are added, it's easy to overlook potential security vulnerabilities. Regular audits help ensure that our security model remains robust and effective.

As we continue to build and refine our agentic AI systems, we'll need to stay vigilant about security and consent. Tomorrow, we'll explore another critical aspect of building robust AI systems, one that will help us take our capabilities to the next level.