0%

MCP Course Week 4: LangGraph

Contents

November 23, 2025

Ai

Llm

Python

Contents

1. LangGraph's Terminology

1.1. State

In LangGraph a state represents the current snapshot of the application.

We define state to be a subclass of either BaseModel from pydantic or TypedDict from typging:

from pydantic import BaseModel
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class State(BaseModel):
    messages: Annotated[list, add_messages]

# or
class State(TypedDict):
    messages: Annotated[list, add_messages]

Here Annotated is a special syntax to provide extra information to the type.

1.2. Node and Edges

  • Nodes are python functions that represent agent logic. They receive the current state as input, do something, and return an updated attribute of the entire state.

    Remark. The state will be upated immediately before it reaches to another node. We explain how the state gets updated in 〈1.3. Reduce Pattern in LangGraph〉.

  • Edges are python functions that determine which node to execute next based on the state, they can be conditional or fixed.

1.3. Reduce Pattern in LangGraph

1.3.1. Example of reducer from react

In old-fasioned react redux a reducer satisfies the interface:

function (state: State, actoin: Action): State

Example:

export default (state: TAppState = initialState, action: AnyAction): TAppState => {
  switch (action.type) {
    case appActions.UPDATE_BLOG: {
      const update = (action as appActions.UpdateBlog).payload;
      return {
        ...state, blog: {
          ...state.blog, ...update
        }
      }
      default:
      return state;
  }
}
1.3.2. Reducer in LangGraph, more explanation on add_message used by "AgentNode" and ToolNdoe

In LangGraph each node is actually a python function that satisfies the following reducer interface:

# psuedo code
function(state: State) -> Partial<State>

  • In LangGraph a node is a reducer that returns "What to Update" (a computed payload) instead of returning the entire updated state.

    We can see one such example in 〈3.3.3. Create Nodes and Bind the Tools〉, the chatbot function.

  • The update of the state of the graph is delegated to LangGraph, but how would LangGraph do the update?

  • LangGraph looks at the annotation:

    class State(BaseModel):
  messages: Annotated[list, add_messages]

    where add_messages is a function (again from langgraph) that appends [message] to the list of messages, and that's how update is done.

  • LangGraph has internally controlled the return type of how Agent returns, add_messages also has internally abstracted how update was executed behind the scene, but under the hood these messages are either:

    • AIMessage from Agent
    • HumanMessage from Agent
    • SystemMessage from Agent
    • ToolMessage from ToolNode

    which we can import from langchain_core.messages.

  • By default without any annotation, the update is simply an assignment, replacing the old value.

    It is not as magical as giving it a str explaning what to do and it does the magical update for us.

2. What is LangSmith? Why do we need it?

Langsmith is a tool bundled with langgraph to trace the execution flow of our graph. To add this tracing:

  1. First let's register an account in https://langsmith.com

  2. You can get the API key and required env variables via

  3. Put those variables into our .env file:

  4. Now when we build a graph with langchain's Tool and StateGraph. It will automatically log down the detail such as the execution of each node and each condition determined by the edges to whether or not to execute next node.

3. Run an Agent by LangGraph

3.1. Define Tools

3.1.1. tool_search (Google Api Wrapper)

Make sure to register an account from https://serper.dev/ (but not https://serpapi.com/), then you should be able to get a free tier API key with a limit of 250 API calls.

Set this API key in .env with key name SERPER_API_KEY:

from langchain.agents import Tool
from langchain_community.utilities import GoogleSerperAPIWrapper

# can skip the keyword argument:
serper = GoogleSerperAPIWrapper(serper_api_key=os.getenv("SERPER_API_KEY"))

tool_search =Tool(
        name="search",
        func=serper.run,
        description="Useful for when you need more information from an online search"
    )

We can test this tool by tool_search.invoke("What is calculus"), and we get:

'Calculus is the mathematical study of continuous change, in the same way that geometry is the study of shape, and algebra is the study of generalizations of ... Calculus is a set of methods and tools used to mathematically investigate and describe how things change, usually by way of cutting up that ... The first thing that calculus does for us it allows us to find the precise exact area of crazy shapes any shape that you want. Calculus is a branch of mathematics that studies the rate of change; it is used to model systems where there is change. Calculus is the study of how things change. It provides a framework for modeling systems in which there is change, and a way to deduce the predictions of such ... Put in the most simple terms, calculus is the study of rates of change. Calculus is one of many mathematics classes taught in high school and college. Calculus, branch of mathematics concerned with instantaneous rates of change and the summation of infinitely many small factors. Calculus is a branch of mathematics that studies continuous change; deals with properties of derivatives and integrals using methods based on the summation of ... Calculus is the mathematical study of change, in the same way that geometry is the study of shape and algebra is the study of operations and their application ... The word Calculus comes from Latin meaning "small stone", because it is like understanding something by looking at small pieces. Differential Calculus cuts ...'
3.1.2. tool_send_email
import sendgrid
import os
from sendgrid.helpers.mail import Mail, Email, To, Content
from agents import Agent, Runner, trace, function_tool

def send_email(text:str):
    sg = sendgrid.SendGridAPIClient(api_key=os.environ.get('SENDGRID_API_KEY'))
    # Change to your verified sender
    from_email = Email("james.lee@wonderbricks.com")
    to_email = To("machingclee@gmail.com")  # Change to your recipient
    content = Content("text/plain", text)
    mail = Mail(from_email, to_email, "Answer to your question", content).get()
    response = sg.client.mail.send.post(request_body=mail)
    print(response.status_code)
    

tool_send_email = Tool(
    name="send_email",
    func=send_email,
    description="useful for when you want to send an email with text as content"
)

3.2. Define get_llm by AzureOpenAI Models

from langchain_openai import AzureChatOpenAI

def get_llm():
    llm = AzureChatOpenAI(
        azure_deployment=os.getenv("AZURE_OPENAI_MODEL"),
        api_version=os.getenv("AZURE_API_VERSION", "2024-10-21"),
        azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
        api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    )
    return llm

# llm = get_llm() to get a model

Here for example we use:

AZURE_OPENAI_API_KEY=XXX
AZURE_OPENAI_ENDPOINT=https://shellscriptmanager.openai.azure.com
AZURE_API_VERSION=2025-01-01-preview
AZURE_OPENAI_MODEL=gpt-4.1-mini

3.3. Define a Graph

A standard LangGraph application goes through the following procedure:

3.3.1. Define the state class
from typing import TypedDict

class State(TypedDict):
    messages: Annotated[list, add_messages]
3.3.2. Start the graph builder
from langgraph.graph import StateGraph,

graph_builder = StateGraph(State)
3.3.3. Create Nodes and Bind the Tools

Recall that we have defined tools tool_search and tool_send_email in 〈3.1. Define Tools〉 and get_llm in 〈3.2. Define get_llm by AzureOpenAI Models〉.

Now we combine these tool as a ToolNode:

from langgraph.prebuilt import ToolNode

llm = get_llm()
tools = [tool_search, tool_send_email]
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    print(state)
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# define nodes
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", ToolNode(tools=tools))

Note that ToolNode is of interface

(State) -> { "messages": List[ToolMessage] }

Whenever our ToolNode gets exectued, LangGraph appends a ToolMessage to our State["messages"], and the next chatbot node will consume the latest message from the ToolNode for the next iteration.

3.3.4. Create Edges

In the following the tools_condition is exactly

  • looking for the model response saying that the finish_reason=="tool_calls"
  • then retrieve the call
  • run the function and provide the results

We have gone through the above process previously in the section Apply the Tools.

from langgraph.graph import START
from langgraph.prebuilt import tools_condition

graph_builder.add_conditional_edges( "chatbot", tools_condition, "tools")
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
3.3.5. Compile the graph
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

LangChain automatically adds the END node for us. Note that dotted lines mean conditional edge. Whether or not to go to the target node depends on the edge condition.

3.4. Execute the Graph via gradio ChatUI

3.4.1. Execution

A raw execution of the graph goes as follows:

result = graph.invoke({"messages": [
    {"role": "user", "content": "Hi, I am James"}
]})
print(result['messages'][-1].content)

Usually we would love to wrap it into a chat function and build a gradio chatUI to test the model:

import gradio as gr

def chat(user_input: str, history):
    result = graph.invoke({"messages": [{"role": "user", "content": user_input}]})
    return result["messages"][-1].content

gr.ChatInterface(chat, type="messages").launch()

3.4.2. The Problem

We see the problem, the state of each of our conversation has not been saved.

This is because we have not persisted the memory of the state by any means for each iteration (in each iteration we run through all of the nodes completely).

It is a simple configuration in langgraph to persist the state of each iteration, this recorder is called a checkpointer:

3.5. Add Memory via checkpointer with Multiple Chat Sessions

Let's define a checkpointer by:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
# Steps 1 and 2
graph_builder = StateGraph(State)

# Step 3
llm = get_llm()
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    print(state)
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", ToolNode(tools=tools))

# Step 4
graph_builder.add_conditional_edges( "chatbot", tools_condition, "tools")
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

# Step 5 (the only extra step)
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}

def chat(user_input: str, history):
    result = graph.invoke({"messages": [{"role": "user", "content": user_input}]}, config=config)
    return result["messages"][-1].content

gr.ChatInterface(chat, type="messages").launch()

Now the states are persisted in memory:

This memory is also identified with a string thread_id, with which we can accomplish adding memory to differnt chat sessions.

But for sure this is not enough as the memory are erased once we reboot our machine. Therefore we go to the next step:

3.6. Persist the Memory in Database

3.6.1. Create checkpointer
3.6.1.1. SQLite

langgraph has native support to sqlite:

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

db_path = "memory.db"
conn = sqlite3.connect(db_path, check_same_thread=False)
checkpointer = SqliteSaver(conn)
3.6.1.2. PostgreSQL

We install necessary libraries for the support of PostgreSQL in langgraph:

uv add langgraph-checkpoint-postgres psycopg_pool psycopg-binary

Similar to previous section we define a checkpointer:

from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

DB_URI = "postgresql://pguser:pguser@localhost:5432/pgdb"
# enable autocommit in the connection arguments
pool = ConnectionPool(DB_URI, kwargs={"autocommit": True})
checkpointer = PostgresSaver(pool)
# create table
checkpointer.setup()
3.6.2. Rerun the graph
3.6.2.1. What has been saved into database?

Same script as before, we pass our checkpointer when compiling the graph:

# Steps 1 and 2
graph_builder = StateGraph(State)

# Step 3
llm = get_llm()
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    print(state)
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", ToolNode(tools=tools))

# Step 4
graph_builder.add_conditional_edges( "chatbot", tools_condition, "tools")
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

# Step 5 (the only extra step)
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}

def chat(user_input: str, history):
    result = graph.invoke({"messages": [{"role": "user", "content": user_input}]}, config=config)
    return result["messages"][-1].content

gr.ChatInterface(chat, type="messages").launch()

Both sqlite and postgresql store the data in a simular format:

In sqlite the results are in linked list format:

where the metadata is our chat data:

3.6.2.2. Retrieve the data

These message can be retrieved from:

history = graph.get_state_history(config)
last_state = next(history)

messages = last_state.values.get("messages", [])
for msg in messages:
    print(f"{msg.type}: {msg.content}")

Which results in:

human: I am james
ai: Hello James! How can I assist you today?
human: what's my name?
ai: Your name is James. How can I help you further?
human: what is my name?
ai: Your name is James. Is there anything else you'd like to know or do?
human: I just want to say hi to you
ai: Hi, James! It's great to hear from you. If you have any questions or just want to chat, I'm here!

4. Agent with Playwright

4.1. Installation

uv run playwright install

4.2. nest_asynci for Python Notebook

Since Jupyter notebook is executed inside an event loop, to run another event loop within an event loop we need:

4.2.1. For Non-Windows User
import nest_asynci
nest_asyncio.apply()
4.2.2. Windows Users

To avoid NotImplementedError and if we would like to run async LangGraph in the notebook, we need to make a small change that is hacky. We need to do this AFTER installing Playwright (prior cells)

  1. Right click in .venv in the File Explorer on the left and select "Find in folder"

  2. Search for

    asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy())

  3. That code should be found in a line of code in a file called kernelapp.py

  4. Comment out the entire else clause that this line is a part of - see the fragment below. Be sure to have the "pass" statement after the ImportError line.

    if sys.platform.startswith("win") and sys.version_info >= (3, 8):
    import asyncio

    try:
        from asyncio import WindowsProactorEventLoopPolicy, WindowsSelectorEventLoopPolicy
    except ImportError:
        pass
        # not affected
    # else:
    #    if type(asyncio.get_event_loop_policy()) is WindowsProactorEventLoopPolicy:
            # WindowsProactorEventLoopPolicy is not compatible with tornado 6
            # fallback to the pre-3.8 default of Selector
            # asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy())

  5. Restart the kernel by pressing the "Restart" button

4.3. Playwright Tools from Langchian

from langchain_community.agent_toolkits import PlayWrightBrowserToolkit
from langchain_community.tools.playwright.utils import create_async_playwright_browser

async_browser =  create_async_playwright_browser(headless=False)  # headful mode
toolkit = PlayWrightBrowserToolkit.from_browser(async_browser=async_browser)
playwright_tools = toolkit.get_tools()

for tool in playwright_tools:
    print(f"{tool.name}")

we shoud get the following tool:

  • click_element
  • navigate_browser
  • previous_webpage
  • extract_text
  • extract_hyperlinks
  • get_elements
  • current_webpage

4.4. Open a Webpage, Scrap data, Get Specfic Info and Send it via Email

4.4.1. Compile the Graph

Recall tool_send_email from the section 〈3.1.2. tool_send_email〉, we combine all tools we need:

all_tools = playwright_tools + [tool_send_email]

Also by using get_llm from 〈3.2. Define get_llm by AzureOpenAI Models〉 we define a graph:

class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

llm = get_llm()
llm_with_tools = llm.bind_tools(all_tools)

# for our main worker node
def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", ToolNode(tools=all_tools))

graph_builder.add_conditional_edges( "chatbot", tools_condition, "tools")
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
display(Image(graph.get_graph().draw_mermaid_png()))

Now we can create a gradio ui to test the agentic result:

config = {"configurable": {"thread_id": "1"}}

async def chat(user_input: str, history):
    result = await graph.ainvoke({"messages": [{"role": "user", "content": user_input}]}, config=config)
    return result["messages"][-1].content

# interface.close() when we want to stop it
interface = gr.ChatInterface(chat, type="messages").launch()
4.4.2. LangGraph Result

5. Multi-Agents Workflow in LangGraph

5.1. State

class State(TypedDict):
    messages: Annotated[List[Any], add_messages]
    success_criteria: str
    feedback_on_work: Optional[str]
    success_criteria_met: bool
    user_input_needed: bool

5.2. Define Agent Nodes with Specific pydantic Response

We define "workder node" and "evaluator node" as follows:

class EvaluatorOutput(BaseModel):
    feedback: str = Field(description="Feedback on the assistant's response")
    success_criteria_met: bool = Field(description="Whether the success criteria have been met")
    user_input_needed: bool = Field(description="True if more input is needed from the user, or clarifications, or the assistant is stuck")

worker_llm = get_llm()
worker_llm_with_tools = worker_llm.bind_tools(tools)

evaluator_llm = get_llm()
evaluator_llm_with_output = evaluator_llm.with_structured_output(EvaluatorOutput)
5.2.1. Worker Node
5.2.1.1. The Node
def worker(state: State) -> Dict[str, Any]:
    system_message = f"""
        You are a helpful assistant that can use tools to complete tasks.
        You keep working on a task until either you have a question or clarification for the user, or the success criteria is met.
        This is the success criteria:
        {state['success_criteria']}
        You should reply either with a question for the user about this assignment, or with your final response.
        If you have a question for the user, you need to reply by clearly stating your question. An example might be:

        Question: please clarify whether you want a summary or a detailed answer

        If you've finished, reply with the final answer, and don't ask a question; simply reply with the answer.
    """
        
    if state.get("feedback_on_work"):
        system_message += f"""
            Previously you thought you completed the assignment, but your reply was rejected because the success criteria was not met.
            Here is the feedback on why this was rejected:
            {state['feedback_on_work']}
            With this feedback, please continue the assignment, ensuring that you meet the success criteria or have a question for the user.
        """
    
    # Add in the system message

    found_system_message = False
    messages = state["messages"]
    for message in messages:
        if isinstance(message, SystemMessage):
            message.content = system_message
            found_system_message = True
    
    if not found_system_message:
        messages = [SystemMessage(content=system_message)] + messages
    
    # Invoke the LLM with tools
    response = worker_llm_with_tools.invoke(messages)
    
    # Return updated state
    return {
        "messages": [response],
    }
5.2.1.2. Next Custom Edge
def worker_router(state: State) -> str:
    last_message = state["messages"][-1]
    
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    else:
        return "evaluator"
5.2.2. Evaluator Node
5.2.2.1. The Node
# helper function to simplify the output for LLM
def format_conversation(messages: List[Any]) -> str:
    conversation = "Conversation history:\n\n"
    for message in messages:
        if isinstance(message, HumanMessage):
            conversation += f"User: {message.content}\n"
        elif isinstance(message, AIMessage):
            text = message.content or "[Tools use]"
            conversation += f"Assistant: {text}\n"
    return conversation

def evaluator(state: State) -> State:
    last_response = state["messages"][-1].content

    system_message = """
        You are an evaluator that determines if a task has been completed successfully by an Assistant.
        Assess the Assistant's last response based on the given criteria. Respond with your feedback, and with your decision on whether the success criteria has been met,
        and whether more input is needed from the user.
        """
        
        user_message = f"""
            You are evaluating a conversation between the User and Assistant. You decide what action to take based on the last response from the Assistant.

            The entire conversation with the assistant, with the user's original request and all replies, is:
            {format_conversation(state['messages'])}

            The success criteria for this assignment is:
            {state['success_criteria']}

            And the final response from the Assistant that you are evaluating is:
            {last_response}

            Respond with your feedback, and decide if the success criteria is met by this response.
            Also, decide if more user input is required, either because the assistant has a question, needs clarification, or seems to be stuck and unable to answer without help.
        """

    if state["feedback_on_work"]:
        user_message += f"Also, note that in a prior attempt from the Assistant, you provided this feedback: {state['feedback_on_work']}\n"
        user_message += "If you're seeing the Assistant repeating the same mistakes, then consider responding that user input is required."
    
    evaluator_messages = [SystemMessage(content=system_message), HumanMessage(content=user_message)]

    eval_result = evaluator_llm_with_output.invoke(evaluator_messages)
    new_state = {
        "messages": [{"role": "assistant", "content": f"Evaluator Feedback on this answer: {eval_result.feedback}"}],
        "feedback_on_work": eval_result.feedback,
        "success_criteria_met": eval_result.success_criteria_met,
        "user_input_needed": eval_result.user_input_needed
    }

    return new_state
5.2.2.2. Next Custom Edge
def route_based_on_evaluation(state: State) -> str:
    if state["success_criteria_met"] or state["user_input_needed"]:
        return "END"
    else:
        return "worker"
5.2.3. Compile the Graph
1graph_builder = StateGraph(State)
2
3# Add nodes
4graph_builder.add_node("worker", worker)
5graph_builder.add_node("tools", ToolNode(tools=tools))
6graph_builder.add_node("evaluator", evaluator)
7
8# Add edges
9graph_builder.add_conditional_edges(
10  "worker", 
11  worker_router, 
12  {"tools": "tools", "evaluator": "evaluator"}
13)

Here line-12 means that

  • If edge function returns tools, then go to tools node (registered in line-5);
  • If edge function returns evaluator, then go to evaluator node (registered in line-6);
14graph_builder.add_edge("tools", "worker")
15graph_builder.add_conditional_edges(
16  "evaluator", 
17  route_based_on_evaluation, 
18  {"worker": "worker", "END": END}
19)
20graph_builder.add_edge(START, "worker")
21
22# Compile the graph
23memory = MemorySaver()
24graph = graph_builder.compile(checkpointer=memory)
25display(Image(graph.get_graph().draw_mermaid_png()))

5.3. Conclusion

  • In LangGraph both Tools and Agents are nothing but node, they can mutate the state in whatever way we want. The mutation is defined by the dictionary returned by each node-function, where the keys are what we want to mutate in the state.

  • By default tool node and agent node must mutate the messages attribute in the state (common design).

  • When there are two agent nodes, both mutate State.messages by appending AIMessage. To distinguish them, we can see the trick in the return of the function evaluator (see 〈5.2.2.1. The Node〉):

    new_state = {
    "messages": [{
      "role": "assistant", 
      "content": f"Evaluator Feedback on this answer: {eval_result.feedback}"
    }],
    ...
}

    We put emphasis on Evaluator Feedback in the content field. Some critical message can also be passed via reassigning value into a specific field of the state (but we still need to keep it in messages if that response should be accumulated as a context).

  • Also note that by setting role = "assistant" in the message, it is the same as

    new_state = {
    "message": [AIMessage(content=f"Evaluator ...")]
}

  • Now we have full control on the flow of how two agents communicate, which is impossible in CrewAI that we study in week 3 of this course (should I skip it entirely as there are too many negative feedbacks on CrewAI).

  • Finally we determine the path of the graph depending on the edge function, which compute the condition based on the state.

6. Special Tools from LangChain

In what follows we will use the following imports:

from langchain_community.agent_toolkits import FileManagementToolkit
from langchain_community.tools.wikipedia.tool import WikipediaQueryRun
from langchain_experimental.tools import PythonREPLTool
from langchain_community.utilities import GoogleSerperAPIWrapper
from langchain_community.utilities.wikipedia import WikipediaAPIWrapper

6.1. Tools

6.1.1. FileManagementToolkit
6.1.1.1. When to use

It Provides file system operations in a sandboxed directory.

6.1.1.2. Tools available

It included (7 total):

  • read_file - Read file contents
  • write_file - Create or overwrite a file
  • copy_file - Copy a file
  • move_file - Move/rename a file
  • delete_file - Delete a file
  • list_directory - List files in a directory
  • file_search - Search for files by pattern
6.1.1.3. Safety

All operations restricted to root_dir="sandbox" directory.

6.1.1.4. Example uses
  • "Read data.txt and summarize it"
  • "Save this result to output.txt"
  • "Find all Python files"
6.1.2. PythonREPLTool
6.1.2.1. When to use

Executes Python code in a REPL environment.

6.1.2.2. Capabilities
  • Run calculations: print(123 * 456)
  • Process data: Parse JSON, manipulate lists
  • Complex logic: Loops, conditionals, algorithms
  • Import libraries: Can use any installed Python package
6.1.2.3. Safefy

High risk - Can execute arbitrary code. Use with caution.

6.1.2.4. Example uses
  • "Calculate the average of these numbers"
  • "Parse this JSON and extract the names"
  • "Generate Fibonacci sequence up to 100"
6.1.3. WikipediaQueryRun
6.1.3.1. When to use

Searches and retrieves information from Wikipedia.

6.1.3.2. When it returns

Text summary of Wikipedia articles (first few paragraphs).

6.1.3.3. Safety

Read-only - external API, safe for production.

6.1.3.4. Example uses
  • "Who was Albert Einstein?"
  • "What is machine learning?"
  • "Tell me about the 2024 Olympics"

6.2. Quick Comparison

ToolPurposeRisk Level
FileManagementToolkitFile operations (read/write/search)Low (sandboxed)
PythonREPLToolExecute Python codeHigh (unrestricted)
WikipediaQueryRunSearch WikipediaLow (read-only)

6.3. Usage in Code

# File operations (sandboxed to "sandbox" directory)
toolkit = FileManagementToolkit(root_dir="sandbox")
file_tools = toolkit.get_tools()  # Returns 7 tools

# Python code execution
python_repl = PythonREPLTool()

# Wikipedia search
wikipedia = WikipediaAPIWrapper()
wiki_tool = WikipediaQueryRun(api_wrapper=wikipedia)

# Combine all
all_tools = file_tools + [python_repl, wiki_tool]

6.4. Example Agent Workflow

User: "Research quantum computing, calculate how many words are in the article, and save to a file"

Agent uses:

  1. WikipediaQueryRun → Get article about quantum computing
  2. PythonREPLTool → Count words: len(article.split())
  3. write_file → Save result to sandbox/word_count.txt

7. Reference