MCP 서버와 함께 에이전트 메모리 사용
이 문서에서는 에이전트 런타임이 표준 인터페이스를 통해 에이전트 메모리에 액세스할 수 있도록 Oracle Agent Memory를 MCP(모델 컨텍스트 프로토콜) 도구로 표시합니다.
학습 내용:
- 스레드 및 메모리 도구를 노출하는 MCP 서버를 구축;
- LangGraph 및 WayFlow에서 동일한 서버를 연결합니다.
참고: 패키지 설정은 에이전트 메모리 시작하기를 참조하십시오. 이 예에 대한 로컬 Oracle AI Database가 필요한 경우 Run Oracle AI Database Locally를 참조하십시오. 이 문서에서는 Oracle DB 풀, 임베더 및 LLM이 이미 구성되어 있다고 가정합니다.
MCP 서버 작성
Oracle Agent 메모리 API를 도구로 노출하는 MCP 서버를 생성합니다.
이 예에서는 공구 표면을 의도적으로 작게 유지합니다.
get_or_create_thread, add_messages, get_messages, add_memory 및 search_memory
예제에서는 모든 도구 결과에 대해 json.dumps(...)를 통해 JSON 문자열을 반환합니다. 이렇게 하면 출력 스키마가 간단해지고 MCP 클라이언트에서 정상적으로 작동합니다.
주: 이 서버 예제를 실행하기 전에 MCP 라이브러리를 별도로 설치하십시오.
import json
import os
from typing import Any
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel
from oracleagentmemory.core.dbschemapolicy import SchemaPolicy
from oracleagentmemory.core.embedders.embedder import Embedder
from oracleagentmemory.core.llms.llm import Llm
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
embedder = Embedder(
model="YOUR_EMBEDDING_MODEL",
api_base="YOUR_EMBEDDING_API_BASE",
api_key="YOUR_EMBEDDING_API_KEY",
)
llm = Llm(
model="YOUR_LLM_MODEL",
api_base="YOUR_LLM_API_BASE",
api_key="YOUR_LLM_API_KEY",
)
db_pool = ... #an oracledb connection or connection pool
class MessageT(BaseModel):
role: str
content: str
id: str | None = None
timestamp: str | None = None
metadata: dict[str, Any] | None = None
def create_server(
memory: OracleAgentMemory | None = None,
host: str | None = None,
port: int | None = None,
path: str | None = None,
) -> FastMCP:
"""Create a FastMCP server exposing Oracle Agent Memory tools."""
agent_memory = memory or OracleAgentMemory(
connection=db_pool,
embedder=embedder,
llm=llm,
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
)
resolved_path = path or os.environ.get("MEMORY_MCP_PATH", "/mcp")
if not resolved_path.startswith("/"):
resolved_path = f"/{resolved_path}"
server = FastMCP(
name="Oracle Agent Memory MCP Server",
host=host or os.environ.get("MEMORY_MCP_HOST", "localhost"),
port=int(port if port is not None else os.environ.get("MEMORY_MCP_PORT", "8003")),
streamable_http_path=resolved_path,
)
@server.tool(
description=(
"Get an existing thread by thread_id, or create one from user_id and optional "
"agent_id."
)
)
def get_or_create_thread(
thread_id: str | None = None,
user_id: str | None = None,
agent_id: str | None = None,
) -> str:
create_kwargs: dict[str, str] = {}
if thread_id is not None:
try:
thread = agent_memory.get_thread(thread_id)
return json.dumps(
{
"thread_id": thread.thread_id,
"user_id": thread.user_id,
"agent_id": thread.agent_id,
}
)
except KeyError:
create_kwargs["thread_id"] = thread_id
if user_id is not None:
create_kwargs["user_id"] = user_id
if agent_id is not None:
create_kwargs["agent_id"] = agent_id
thread = agent_memory.create_thread(**create_kwargs)
return json.dumps(
{
"thread_id": thread.thread_id,
"user_id": thread.user_id,
"agent_id": thread.agent_id,
}
)
@server.tool(
description=(
"Add durable memory content, optionally scoped by user_id, agent_id, and "
"thread_id."
)
)
def add_memory(
content: str,
user_id: str | None = None,
agent_id: str | None = None,
thread_id: str | None = None,
) -> str:
add_kwargs: dict[str, str] = {}
if user_id is not None:
add_kwargs["user_id"] = user_id
if agent_id is not None:
add_kwargs["agent_id"] = agent_id
if thread_id is not None:
add_kwargs["thread_id"] = thread_id
return json.dumps({"memory_id": agent_memory.add_memory(content, **add_kwargs)})
@server.tool(
description="Add messages to an existing thread using messages and thread_id."
)
def add_messages(messages: list[MessageT], thread_id: str) -> str:
thread = agent_memory.get_thread(thread_id)
payload = [message.model_dump(exclude_none=True) for message in messages]
return json.dumps({"message_ids": thread.add_messages(payload)})
@server.tool(description="Get messages from an existing thread using thread_id.")
def get_messages(thread_id: str) -> str:
thread = agent_memory.get_thread(thread_id)
return json.dumps(
{
"messages": [
{
"id": getattr(message, "id", None),
"role": message.role,
"content": message.content,
"timestamp": message.timestamp,
"metadata": message.metadata,
}
for message in thread.get_messages()
]
}
)
@server.tool(
description=(
"Search Oracle Agent Memory for durable memory and thread content. "
"Pass user_id directly, or pass thread_id so the server can resolve the user scope."
)
)
def search_memory(
query: str,
user_id: str | None = None,
agent_id: str | None = None,
thread_id: str | None = None,
) -> str:
if user_id is not None:
resolved_user_id = user_id
resolved_agent_id = agent_id
elif thread_id is not None:
thread = agent_memory.get_thread(thread_id)
if thread.user_id is None:
raise ValueError(
f"Thread `{thread_id}` is not associated with a user_id, so "
"search_memory cannot build a valid OracleAgentMemory search scope."
)
resolved_user_id = thread.user_id
resolved_agent_id = agent_id if agent_id is not None else thread.agent_id
else:
raise ValueError("search_memory requires either `user_id` or `thread_id`.")
search_kwargs: dict[str, Any] = {"query": query, "user_id": resolved_user_id}
if resolved_agent_id is not None:
search_kwargs["agent_id"] = resolved_agent_id
search_kwargs["exact_agent_match"] = True
if thread_id is not None:
search_kwargs["thread_id"] = thread_id
search_kwargs["exact_thread_match"] = True
results = agent_memory.search(**search_kwargs)
return json.dumps(
{
"results": [
{
"id": result.id,
"content": result.content,
"record_type": result.record.record_type,
"user_id": result.record.user_id,
"agent_id": result.record.agent_id,
"thread_id": result.record.thread_id,
}
for result in results
]
}
)
return serverMCP 서버 실행
기본적으로 서버는 http://localhost:8003/mcp에서 수신합니다. 필요한 경우 바인드 주소를 MEMORY_MCP_HOST, MEMORY_MCP_PORT 및 MEMORY_MCP_PATH로 재정의합니다.
def main() -> None:
server = create_server()
server.run(transport="streamable-http")
if __name__ == "__main__":
main()LangGraph에서 서버 사용
LangGraph는 langchain-mcp-adapters를 통해 MCP 도구를 사용할 수 있습니다. 아래 에이전트는 streamable-http를 통해 메모리 서버에 접속하여 모델이 Oracle Agent 메모리 툴을 직접 호출할 수 있도록 합니다.
주: 이 클라이언트 예제에 대해 langchain-mcp-adapters를 설치합니다. LangGraph MCP 도구는 ainvoke()와 같은 비동기 메소드로 에이전트를 실행해야 합니다.
LangGraph 클라이언트 구성
import os
from datetime import timedelta
import anyio
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
mcp_url = os.environ.get("MEMORY_MCP_URL", "http://localhost:8003/mcp")
langgraph_llm = ChatOpenAI(
model="gpt-4.1-mini",
api_key="YOUR_OPENAI_API_KEY",
)
async def build_langgraph_agent():
memory_tools = await MultiServerMCPClient(
{
"memory": {
"transport": "streamable_http",
"url": mcp_url,
"timeout": timedelta(seconds=30),
"sse_read_timeout": timedelta(seconds=30),
}
}
).get_tools()
return create_agent(
model=langgraph_llm,
tools=memory_tools,
system_prompt=(
"You are an assistant using Oracle Agent Memory through MCP. "
"Create threads before writing messages, use add_memory for durable facts, "
"and call search_memory when the user asks about prior context."
),
)MCP 서버에 대해 LangGraph 에이전트 실행
async def run_langgraph_agent() -> None:
agent = await build_langgraph_agent()
first_turn = await agent.ainvoke(
{
"messages": [
HumanMessage(
content=(
"Use the memory MCP tools to create thread `mcp_demo_thread` for user "
"`user_123`, add the durable memory `The user likes orange juice with "
"breakfast.`, and confirm when the memory is stored."
)
)
]
}
)
print(first_turn["messages"][-1].content)
second_turn = await agent.ainvoke(
{
"messages": [
HumanMessage(
content=(
"Search memory for `orange juice` in thread `mcp_demo_thread` and tell "
"me what Oracle Agent Memory returned."
)
)
]
}
)
print(second_turn["messages"][-1].content)출력:
The durable memory has been stored successfully:
-Thread ID: `mcp_demo_thread`
-User ID: `user_123`
The search returned matching memory records for:
"The user likes orange juice with breakfast."WayFlow에서 서버 사용
WayFlow는 MCPToolBox를 통해 StreamableHTTPTransport를 사용하여 동일한 MCP 서버를 사용할 수 있습니다. API 세부정보는 WayFlow MCPToolBox 문서를 참조하십시오.
주: 이 클라이언트 예제에 대해 wayflowcore를 설치합니다.
WayFlow 클라이언트 구성
import os
from wayflowcore.agent import Agent
from wayflowcore.mcp import MCPToolBox, StreamableHTTPTransport, enable_mcp_without_auth
from wayflowcore.models import OpenAICompatibleModel
mcp_url = os.environ.get("MEMORY_MCP_URL", "http://localhost:8003/mcp")
wayflow_llm = OpenAICompatibleModel(
model_id="gpt-4.1-mini",
base_url="YOUR_OPENAI_API_BASE",
api_key="YOUR_OPENAI_API_KEY",
)
enable_mcp_without_auth()
memory_tools = MCPToolBox(client_transport=StreamableHTTPTransport(url=mcp_url))
agent = Agent(
llm=wayflow_llm,
agent_id="memory_mcp_agent",
custom_instruction=(
"You are an assistant using Oracle Agent Memory through MCP. "
"Create threads before writing messages, use add_memory for durable facts, "
"and call search_memory when the user asks about prior context."
),
tools=[memory_tools],
)MCP 서버에 대해 WayFlow 에이전트 실행
first_session = agent.start_conversation()
first_session.append_user_message(
"Use the memory MCP tools to create thread `mcp_demo_thread` for user `user_123`, "
"add the durable memory `The user likes orange juice with breakfast.`, and confirm "
"when the memory is stored."
)
first_session.execute()
print(first_session.get_last_message().content)
second_session = agent.start_conversation()
second_session.append_user_message(
"Search memory for `orange juice` in thread `mcp_demo_thread` and tell me what "
"Oracle Agent Memory returned."
)
second_session.execute()
print(second_session.get_last_message().content)출력:
The durable memory has been stored successfully:
-Thread ID: `mcp_demo_thread`
-User ID: `user_123`
The memory search returned matching records for:
"The user likes orange juice with breakfast."보안 및 배치 노트
예제 서버는 의도적으로 열려 있고 로컬 전용입니다. 운용 용도로 인증 및 전송 보안을 추가하고, 각 호출자가 액세스할 수 있는 사용자 또는 에이전트를 제한하고, add_memory 또는 add_messages와 같은 쓰기 작업에 대해 확인 또는 승인 게이트를 추가하는 것을 고려합니다.
자세한 내용은 보안 고려 사항을 참조하십시오.
결론
이 문서에서는 MCP를 통해 Oracle Agent 메모리를 노출하고 LangGraph 및 WayFlow에서 동일한 도구 모음을 재사용하는 방법을 배웠습니다.
참고: 에이전트 워크플로우에 필요한 경우에만 메모리 도구를 추가할 수 있습니다. 빠른 참조 코드 샘플을 참조하십시오. 배포 환경 및 운영 제약 조건에 맞게 서버를 조정하려면 Run Oracle AI Database Locally를 참조하십시오.
전체 코드
MCP 서버
#Copyright © 2026 Oracle and/or its affiliates.
#This software is under the Apache License 2.0
#(LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License
#(UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option.
#Oracle Agent Memory Code Example - Integration with MCP Server
#--------------------------------------------------------------
#How to use:
#Create a new Python virtual environment and install the latest oracleagentmemory version.
#You can now run the script
#1. As a Python file:
#```bash
#python integration_with_mcp_server.py
#```
#2. As a Notebook (in VSCode):
#When viewing the file,
#- press the keys Ctrl + Enter to run the selected cell
#- or Shift + Enter to run the selected cell and move to the cell below
##Build the MCP server
#%%
import json
import os
from typing import Any
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel
from oracleagentmemory.core.dbschemapolicy import SchemaPolicy
from oracleagentmemory.core.embedders.embedder import Embedder
from oracleagentmemory.core.llms.llm import Llm
from oracleagentmemory.core.oracleagentmemory import OracleAgentMemory
embedder = Embedder(
model="YOUR_EMBEDDING_MODEL",
api_base="YOUR_EMBEDDING_API_BASE",
api_key="YOUR_EMBEDDING_API_KEY",
)
llm = Llm(
model="YOUR_LLM_MODEL",
api_base="YOUR_LLM_API_BASE",
api_key="YOUR_LLM_API_KEY",
)
db_pool = ... #an oracledb connection or connection pool
class MessageT(BaseModel):
role: str
content: str
id: str | None = None
timestamp: str | None = None
metadata: dict[str, Any] | None = None
def create_server(
memory: OracleAgentMemory | None = None,
host: str | None = None,
port: int | None = None,
path: str | None = None,
) -> FastMCP:
"""Create a FastMCP server exposing Oracle Agent Memory tools."""
agent_memory = memory or OracleAgentMemory(
connection=db_pool,
embedder=embedder,
llm=llm,
schema_policy=SchemaPolicy.CREATE_IF_NECESSARY,
)
resolved_path = path or os.environ.get("MEMORY_MCP_PATH", "/mcp")
if not resolved_path.startswith("/"):
resolved_path = f"/{resolved_path}"
server = FastMCP(
name="Oracle Agent Memory MCP Server",
host=host or os.environ.get("MEMORY_MCP_HOST", "localhost"),
port=int(port if port is not None else os.environ.get("MEMORY_MCP_PORT", "8003")),
streamable_http_path=resolved_path,
)
@server.tool(
description=(
"Get an existing thread by thread_id, or create one from user_id and optional "
"agent_id."
)
)
def get_or_create_thread(
thread_id: str | None = None,
user_id: str | None = None,
agent_id: str | None = None,
) -> str:
create_kwargs: dict[str, str] = {}
if thread_id is not None:
try:
thread = agent_memory.get_thread(thread_id)
return json.dumps(
{
"thread_id": thread.thread_id,
"user_id": thread.user_id,
"agent_id": thread.agent_id,
}
)
except KeyError:
create_kwargs["thread_id"] = thread_id
if user_id is not None:
create_kwargs["user_id"] = user_id
if agent_id is not None:
create_kwargs["agent_id"] = agent_id
thread = agent_memory.create_thread(**create_kwargs)
return json.dumps(
{
"thread_id": thread.thread_id,
"user_id": thread.user_id,
"agent_id": thread.agent_id,
}
)
@server.tool(
description=(
"Add durable memory content, optionally scoped by user_id, agent_id, and "
"thread_id."
)
)
def add_memory(
content: str,
user_id: str | None = None,
agent_id: str | None = None,
thread_id: str | None = None,
) -> str:
add_kwargs: dict[str, str] = {}
if user_id is not None:
add_kwargs["user_id"] = user_id
if agent_id is not None:
add_kwargs["agent_id"] = agent_id
if thread_id is not None:
add_kwargs["thread_id"] = thread_id
return json.dumps({"memory_id": agent_memory.add_memory(content, **add_kwargs)})
@server.tool(
description="Add messages to an existing thread using messages and thread_id."
)
def add_messages(messages: list[MessageT], thread_id: str) -> str:
thread = agent_memory.get_thread(thread_id)
payload = [message.model_dump(exclude_none=True) for message in messages]
return json.dumps({"message_ids": thread.add_messages(payload)})
@server.tool(description="Get messages from an existing thread using thread_id.")
def get_messages(thread_id: str) -> str:
thread = agent_memory.get_thread(thread_id)
return json.dumps(
{
"messages": [
{
"id": getattr(message, "id", None),
"role": message.role,
"content": message.content,
"timestamp": message.timestamp,
"metadata": message.metadata,
}
for message in thread.get_messages()
]
}
)
@server.tool(
description=(
"Search Oracle Agent Memory for durable memory and thread content. "
"Pass user_id directly, or pass thread_id so the server can resolve the user scope."
)
)
def search_memory(
query: str,
user_id: str | None = None,
agent_id: str | None = None,
thread_id: str | None = None,
) -> str:
if user_id is not None:
resolved_user_id = user_id
resolved_agent_id = agent_id
elif thread_id is not None:
thread = agent_memory.get_thread(thread_id)
if thread.user_id is None:
raise ValueError(
f"Thread `{thread_id}` is not associated with a user_id, so "
"search_memory cannot build a valid OracleAgentMemory search scope."
)
resolved_user_id = thread.user_id
resolved_agent_id = agent_id if agent_id is not None else thread.agent_id
else:
raise ValueError("search_memory requires either `user_id` or `thread_id`.")
search_kwargs: dict[str, Any] = {"query": query, "user_id": resolved_user_id}
if resolved_agent_id is not None:
search_kwargs["agent_id"] = resolved_agent_id
search_kwargs["exact_agent_match"] = True
if thread_id is not None:
search_kwargs["thread_id"] = thread_id
search_kwargs["exact_thread_match"] = True
results = agent_memory.search(**search_kwargs)
return json.dumps(
{
"results": [
{
"id": result.id,
"content": result.content,
"record_type": result.record.record_type,
"user_id": result.record.user_id,
"agent_id": result.record.agent_id,
"thread_id": result.record.thread_id,
}
for result in results
]
}
)
return server
##Run the MCP server
#%%
def main() -> None:
server = create_server()
server.run(transport="streamable-http")
if __name__ == "__main__":
main()LangGraph 클라이언트
#Copyright © 2026 Oracle and/or its affiliates.
#This software is under the Apache License 2.0
#(LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License
#(UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option.
#Oracle Agent Memory Code Example - LangGraph MCP Client
#-------------------------------------------------------
#How to use:
#Create a new Python virtual environment and install the latest oracleagentmemory version.
#You can now run the script
#1. As a Python file:
#```bash
#python integration_with_mcp_langgraph.py
#```
#2. As a Notebook (in VSCode):
#When viewing the file,
#- press the keys Ctrl + Enter to run the selected cell
#- or Shift + Enter to run the selected cell and move to the cell below
##Connect LangGraph to the MCP server
#%%
import os
from datetime import timedelta
import anyio
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
mcp_url = os.environ.get("MEMORY_MCP_URL", "http://localhost:8003/mcp")
langgraph_llm = ChatOpenAI(
model="gpt-4.1-mini",
api_key="YOUR_OPENAI_API_KEY",
)
async def build_langgraph_agent():
memory_tools = await MultiServerMCPClient(
{
"memory": {
"transport": "streamable_http",
"url": mcp_url,
"timeout": timedelta(seconds=30),
"sse_read_timeout": timedelta(seconds=30),
}
}
).get_tools()
return create_agent(
model=langgraph_llm,
tools=memory_tools,
system_prompt=(
"You are an assistant using Oracle Agent Memory through MCP. "
"Create threads before writing messages, use add_memory for durable facts, "
"and call search_memory when the user asks about prior context."
),
)
##Use the MCP server from LangGraph
#%%
async def run_langgraph_agent() -> None:
agent = await build_langgraph_agent()
first_turn = await agent.ainvoke(
{
"messages": [
HumanMessage(
content=(
"Use the memory MCP tools to create thread `mcp_demo_thread` for user "
"`user_123`, add the durable memory `The user likes orange juice with "
"breakfast.`, and confirm when the memory is stored."
)
)
]
}
)
print(first_turn["messages"][-1].content)
#The durable memory has been stored successfully:
#- Thread ID: `mcp_demo_thread`
#- User ID: `user_123`
second_turn = await agent.ainvoke(
{
"messages": [
HumanMessage(
content=(
"Search memory for `orange juice` in thread `mcp_demo_thread` and tell "
"me what Oracle Agent Memory returned."
)
)
]
}
)
print(second_turn["messages"][-1].content)
#The search returned matching memory records for:
#"The user likes orange juice with breakfast."
anyio.run(run_langgraph_agent)WayFlow 클라이언트
#Copyright © 2026 Oracle and/or its affiliates.
#This software is under the Apache License 2.0
#(LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License
#(UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option.
#Oracle Agent Memory Code Example - WayFlow MCP Client
#-----------------------------------------------------
#How to use:
#Create a new Python virtual environment and install the latest oracleagentmemory version.
#You can now run the script
#1. As a Python file:
#```bash
#python integration_with_mcp_wayflow.py
#```
#2. As a Notebook (in VSCode):
#When viewing the file,
#- press the keys Ctrl + Enter to run the selected cell
#- or Shift + Enter to run the selected cell and move to the cell below
##Connect WayFlow to the MCP server
#%%
import os
from wayflowcore.agent import Agent
from wayflowcore.mcp import MCPToolBox, StreamableHTTPTransport, enable_mcp_without_auth
from wayflowcore.models import OpenAICompatibleModel
mcp_url = os.environ.get("MEMORY_MCP_URL", "http://localhost:8003/mcp")
wayflow_llm = OpenAICompatibleModel(
model_id="gpt-4.1-mini",
base_url="YOUR_OPENAI_API_BASE",
api_key="YOUR_OPENAI_API_KEY",
)
enable_mcp_without_auth()
memory_tools = MCPToolBox(client_transport=StreamableHTTPTransport(url=mcp_url))
agent = Agent(
llm=wayflow_llm,
agent_id="memory_mcp_agent",
custom_instruction=(
"You are an assistant using Oracle Agent Memory through MCP. "
"Create threads before writing messages, use add_memory for durable facts, "
"and call search_memory when the user asks about prior context."
),
tools=[memory_tools],
)
##Use the MCP server from WayFlow
#%%
first_session = agent.start_conversation()
first_session.append_user_message(
"Use the memory MCP tools to create thread `mcp_demo_thread` for user `user_123`, "
"add the durable memory `The user likes orange juice with breakfast.`, and confirm "
"when the memory is stored."
)
first_session.execute()
print(first_session.get_last_message().content)
#The durable memory has been stored successfully:
#- Thread ID: `mcp_demo_thread`
#- User ID: `user_123`
second_session = agent.start_conversation()
second_session.append_user_message(
"Search memory for `orange juice` in thread `mcp_demo_thread` and tell me what "
"Oracle Agent Memory returned."
)
second_session.execute()
print(second_session.get_last_message().content)
#The memory search returned matching records for:
#"The user likes orange juice with breakfast."