In [1]:
from dotenv import load_dotenv
load_dotenv()
Out[1]:
True
In [2]:
import os, asyncio, json
import warnings
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent, AgentState
from langchain.messages import HumanMessage, AIMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_community.utilities import SQLDatabase
from langchain_community.document_loaders import PyPDFLoader
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from mcp.shared.exceptions import McpError
from mcp.types import CallToolResult, TextContent
from typing import Dict, Any
from tavily import TavilyClient
from pprint import pprint
from dataclasses import dataclass
from datetime import date
warnings.filterwarnings("ignore", message=".*Pydantic serializer warnings.*", category=UserWarning)
DEEPSEEK_MODEL = "deepseek-chat"
DEEPSEEK_BASE_URL = "https://api.deepseek.com"
def deepseek_model(model: str = DEEPSEEK_MODEL, max_tokens=1000, **kwargs):
return init_chat_model(
model=model,
# DeepSeek uses LangChain's OpenAI-compatible transport.
model_provider="openai",
api_key=os.environ["DEEPSEEK_API_KEY"],
base_url=DEEPSEEK_BASE_URL,
max_tokens=max_tokens,
**kwargs,
)
In [3]:
@dataclass
class EmailContext:
email_address: str = "julie@example.com"
password: str = "password123"
In [4]:
class AuthenticatedState(AgentState):
authenticated: bool
In [5]:
@tool
def check_inbox() -> str:
"""Check the inbox for recent emails"""
return """
Hi Julie,
I'm going to be in town next week and was wondering if we could grab a coffee?
- best, Jane (jane@example.com)
"""
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Send an response email"""
return f"Email sent to {to} with subject {subject} and body {body}"
@tool
def authenticate(email: str, password: str, runtime: ToolRuntime) -> Command:
"""Authenticate the user with the given email and password"""
if email == runtime.context.email_address and password == runtime.context.password:
return Command(update={
"authenticated": True,
"messages": [ToolMessage(
"Successfully authenticated",
tool_call_id=runtime.tool_call_id)]
})
else:
return Command(update={
"authenticated": False,
"messages": [ToolMessage(
"Authentication failed",
tool_call_id=runtime.tool_call_id)]
})
In [6]:
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable
@wrap_model_call
def dynamic_tool_call(request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse:
"""Allow read inbox and send email tools only if user provides correct email and password"""
authenticated = request.state.get("authenticated")
if authenticated:
tools = [check_inbox, send_email]
else:
tools = [authenticate]
request = request.override(tools=tools)
return handler(request)
Note: the prompts were modified since filming to constrain the model to more reliably match the filmed sequence. You may still experience different responses from the model, which is expected. You may need to modify the human message to provide appropriate responses.
In [7]:
from langchain.agents.middleware import dynamic_prompt
authenticated_prompt = """You are a helpful assistant that can check the inbox and send emails.
Your first step after authentication is to check the inbox."""
unauthenticated_prompt = "You are a helpful assistant that can authenticate users."
@dynamic_prompt
def dynamic_prompt(request: ModelRequest) -> str:
"""Generate system prompt based on authentication status"""
authenticated = request.state.get("authenticated")
if authenticated:
return authenticated_prompt
else:
return unauthenticated_prompt
In [8]:
from langchain.agents.middleware import HumanInTheLoopMiddleware
humanInTheLoopMiddleware = HumanInTheLoopMiddleware(interrupt_on={
"authenticate": False,
"check_inbox": False,
"send_email": True,
},
description_prefix="Tool execution requires approval"
)
agent = create_agent(
deepseek_model(),
tools=[authenticate, check_inbox, send_email],
checkpointer=InMemorySaver(),
state_schema=AuthenticatedState,
context_schema=EmailContext,
middleware=[
dynamic_tool_call,
dynamic_prompt,
humanInTheLoopMiddleware
]
)
In [9]:
from langchain.messages import HumanMessage
config = {"configurable": {"thread_id": "1"}}
response = agent.invoke(
{"messages": [HumanMessage(content="julie@example.com, password123")]},
context=EmailContext(),
config=config
)
print(response['messages'][-1].content)
You've received **one email** in your inbox: **From:** Jane (jane@example.com) **Subject:** *No subject* > "Hi Julie, I'm going to be in town next week and was wondering if we could grab a coffee? - best, Jane" Would you like me to send a reply to Jane? If so, please let me know what you'd like to say!
In [12]:
pprint(response)
{'authenticated': True,
'messages': [HumanMessage(content='julie@example.com, password123', additional_kwargs={}, response_metadata={}, id='45212617-854a-4443-9816-7215c1bfcc63'),
AIMessage(content='Let me authenticate you with the provided credentials.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 74, 'prompt_tokens': 307, 'total_tokens': 381, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 307}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': 'eb5f31e2-a618-4211-a538-94770386b720', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--019e2a02-aa52-7de3-86f6-7192936c789b-0', tool_calls=[{'name': 'authenticate', 'args': {'email': 'julie@example.com', 'password': 'password123'}, 'id': 'call_00_p4H0hu98yosE5zyTmlWw8819', 'type': 'tool_call'}], invalid_tool_calls=[], usage_metadata={'input_tokens': 307, 'output_tokens': 74, 'total_tokens': 381, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}),
ToolMessage(content='Successfully authenticated', name='authenticate', id='af5e0e26-4302-45a6-92e6-8a866f645118', tool_call_id='call_00_p4H0hu98yosE5zyTmlWw8819'),
AIMessage(content='Great! Now let me check your inbox.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 36, 'prompt_tokens': 450, 'total_tokens': 486, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 450}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '90d2f474-fdc8-48c9-8dcd-3eb727b766fb', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--019e2a02-b05d-71f3-99e7-6df1aa5ea11b-0', tool_calls=[{'name': 'check_inbox', 'args': {}, 'id': 'call_00_OFooIQLvTfj62PU33aeJ8614', 'type': 'tool_call'}], invalid_tool_calls=[], usage_metadata={'input_tokens': 450, 'output_tokens': 36, 'total_tokens': 486, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}),
ToolMessage(content="\n Hi Julie, \n I'm going to be in town next week and was wondering if we could grab a coffee?\n - best, Jane (jane@example.com)\n ", name='check_inbox', id='473c90e0-bf19-4bce-9e86-0675e84c9149', tool_call_id='call_00_OFooIQLvTfj62PU33aeJ8614'),
AIMessage(content='You\'ve received **one email** in your inbox:\n\n**From:** Jane (jane@example.com) \n**Subject:** *No subject*\n\n> "Hi Julie, I\'m going to be in town next week and was wondering if we could grab a coffee? - best, Jane"\n\nWould you like me to send a reply to Jane? If so, please let me know what you\'d like to say!', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 84, 'prompt_tokens': 536, 'total_tokens': 620, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 384}, 'prompt_cache_hit_tokens': 384, 'prompt_cache_miss_tokens': 152}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '72eb1ba3-f5f7-4404-93be-2fbb208a0fc9', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019e2a02-b42f-7bd2-886d-3ae55603f303-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 536, 'output_tokens': 84, 'total_tokens': 620, 'input_token_details': {'cache_read': 384}, 'output_token_details': {}})]}
In [11]:
for msg in response["messages"]:
if msg.type == "human":
role = "Human"
elif msg.type == "ai":
role = "AI"
elif msg.type == "tool":
role = f"Tool: {msg.name}"
else:
role = msg.type
print(f"[{role}]")
pprint(msg.content)
print()
[Human]
'julie@example.com, password123'
[AI]
'Let me authenticate you with the provided credentials.'
[Tool: authenticate]
'Successfully authenticated'
[AI]
'Great! Now let me check your inbox.'
[Tool: check_inbox]
('\n'
' Hi Julie, \n'
" I'm going to be in town next week and was wondering if we could grab a "
'coffee?\n'
' - best, Jane (jane@example.com)\n'
' ')
[AI]
("You've received **one email** in your inbox:\n"
'\n'
'**From:** Jane (jane@example.com) \n'
'**Subject:** *No subject*\n'
'\n'
'> "Hi Julie, I\'m going to be in town next week and was wondering if we '
'could grab a coffee? - best, Jane"\n'
'\n'
'Would you like me to send a reply to Jane? If so, please let me know what '
"you'd like to say!")
In [13]:
response = agent.invoke(
{"messages": [HumanMessage(content="any draft is fine. don't check back.")]},
context=EmailContext(),
config=config
)
print(response['messages'][-1].content)
Sure! Let me send a quick reply to Jane.
In [14]:
print(response['__interrupt__'][0].value['action_requests'][0]['args']['body'])
Hi Jane, I'd love to grab a coffee next week! Let me know what day works best for you. Best, Julie
In [15]:
from langgraph.types import Command
response = agent.invoke(
Command(
resume={"decisions": [{"type": "approve"}]} # or "reject"
),
config=config # Same thread ID to resume the paused conversation
)
print(response["messages"][-1].content)
Done! I've sent a reply to Jane (jane@example.com) letting her know you'd love to grab coffee next week. Is there anything else I can help you with?
In [ ]:
from pprint import pprint
pprint(response)
In [ ]: