How to connect MCP tools to Assistants

Download Python Script

Python script/notebook for this guide.

MCP how-to script

Prerequisites

This guide assumes familiarity with:

• Tools

• Building Assistants with Tools

Model Context Protocol (MCP) is an open protocol that standardizes how applications provide context to LLMs. You can use an MCP server to provide a consistent tool interface to your agents and flows, without having to create custom adapters for different APIs.

Tip

See the Oracle MCP Server Repository to explore examples of reference implementations of MCP servers for managing and interacting with Oracle products.

In this guide, you will learn how to:

• Create a simple MCP Server (in a separate Python file)

• Connect an Agent to an MCP Server (including how to export/load via Agent Spec, and run it)

• Connect a Flow to an MCP Server (including export/load/run)

Important

This guide does not aim at explaining how to make secure MCP servers, but instead mainly aims at showing how to connect to one. You should ensure that your MCP server configurations are secure, and only connect to trusted external MCP servers.

Prerequisite: Setup a simple MCP Server

First, let’s see how to create and start a simple MCP server exposing a couple of tools.

Note

You should copy the following server code and run it in a separate Python process.

from mcp.server.fastmcp import FastMCP

PAYSLIPS = [
    {
        "Amount": 7612,
        "Currency": "USD",
        "PeriodStartDate": "2025/05/15",
        "PeriodEndDate": "2025/06/15",
        "PaymentDate": "",
        "DocumentId": 2,
        "PersonId": 2,
    },
    {
        "Amount": 5000,
        "Currency": "CHF",
        "PeriodStartDate": "2024/05/01",
        "PeriodEndDate": "2024/06/01",
        "PaymentDate": "2024/05/15",
        "DocumentId": 1,
        "PersonId": 1,
    },
    {
        "Amount": 10000,
        "Currency": "EUR",
        "PeriodStartDate": "2025/06/15",
        "PeriodEndDate": "2025/10/15",
        "PaymentDate": "",
        "DocumentsId": 3,
        "PersonId": 3,
    },
]

def create_server(host: str, port: int):
    """Create and configure the MCP server"""
    server = FastMCP(
        name="Example MCP Server",
        instructions="A MCP Server.",
        host=host,
        port=port,
    )

    @server.tool(description="Return session details for the current user")
    def get_user_session():
        return {
            "PersonId": "1",
            "Username": "Bob.b",
            "DisplayName": "Bob B",
        }

    @server.tool(description="Return payslip details for a given PersonId")
    def get_payslips(PersonId: int):
        return [payslip for payslip in PAYSLIPS if payslip["PersonId"] == int(PersonId)]

    return server


def start_mcp_server() -> str:
    host: str = "localhost"
    port: int = 8080
    server = create_server(host=host, port=port)
    server.run(transport="sse")

    return f"http://{host}:{port}/sse"

# mcp_server_url = start_mcp_server() # <--- Move the code above to a separate file then uncomment

This MCP server exposes two example tools: get_user_session and get_payslips. Once started, it will be available at (by default): http://localhost:8080/sse.

Note

When choosing a transport for MCP:

• Use Stdio when launching and communicating with an MCP server as a local subprocess on the same machine as the client.

• Use Streamable HTTP when connecting to a remote MCP server.

For more information, visit https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#stdio

Connecting an Agent to the MCP Server

You can now connect an agent to this running MCP server.

Add imports and configure an LLM

Start by importing the necessary packages for this guide:

from wayflowcore.executors.executionstatus import FinishedStatus, UserMessageRequestStatus
from wayflowcore.agent import Agent
from wayflowcore.mcp import MCPTool, MCPToolBox, SSETransport, enable_mcp_without_auth
from wayflowcore.flow import Flow
from wayflowcore.steps import ToolExecutionStep

mcp_server_url = f"http://localhost:8080/sse" # change to your own URL
# We will see below how to connect a specific tool to an assistant, e.g.
MCP_TOOL_NAME = "get_user_session"
# And see how to build an agent that can answer questions, e.g.
USER_QUERY = "What was the payment date of the last payslip for the current user?"

WayFlow supports several LLM API providers. Select an LLM from the options below:

OCI GenAIvLLMOllama

from wayflowcore.models import OCIGenAIModel

if __name__ == "__main__":

    llm = OCIGenAIModel(
        model_id="provider.model-id",
        service_endpoint="https://url-to-service-endpoint.com",
        compartment_id="compartment-id",
        auth_type="API_KEY",
    )

Build the Agent

Agents can connect to MCP tools by either using a MCPToolBox or a MCPTool. Here you will use the toolbox (see the section on Flows to see how to use the MCPTool).

enable_mcp_without_auth() # <--- See https://modelcontextprotocol.io/specification/2025-06-18/basic/authorization#security-considerations
mcp_client = SSETransport(url=mcp_server_url)
mcp_toolbox = MCPToolBox(client_transport=mcp_client)

assistant = Agent(
    llm=llm,
    tools=[mcp_toolbox]
)

Specify the transport to use to handle the connection to the server and create the toolbox. You can then equip an agent with the toolbox similarly to tools.

Note

enable_mcp_without_auth() disables authorization for local/testing only—do not use in production.

Running the Agent

You can now run the agent in a simple conversation:

# With a linear conversation
conversation = assistant.start_conversation()

conversation.append_user_message(USER_QUERY)
status = conversation.execute()
if isinstance(status, UserMessageRequestStatus):
    assistant_reply = conversation.get_last_message()
    print(f"---\nAssistant >>> {assistant_reply.content}\n---")
else:
    print(f"Invalid execution status, expected UserMessageRequestStatus, received {type(status)}")

# then continue the conversation

Alternatively, run the agent interactively in a command-line loop:

def run_agent_in_command_line(assistant: Agent):
    inputs = {}
    conversation = assistant.start_conversation(inputs)

    while True:
        status = conversation.execute()
        if isinstance(status, FinishedStatus):
            break
        assistant_reply = conversation.get_last_message()
        if assistant_reply is not None:
            print("\nAssistant >>>", assistant_reply.content)
        user_input = input("\nUser >>> ")
        conversation.append_user_message(user_input)

# run_agent_in_command_line(assistant)
# ^ uncomment and execute

Connecting a Flow to the MCP Server

You can also use MCP tools in a Flow by using the MCPTool in a ToolExecutionStep.

Build the Flow

Create the flow using the MCP tool:

mcp_tool = MCPTool(
    name=MCP_TOOL_NAME,
    client_transport=mcp_client
)

assistant = Flow.from_steps([
    ToolExecutionStep(name="mcp_tool_step", tool=mcp_tool)
])

Here you specify the client transport as with the MCP ToolBox, as well as the name of the specific tool you want to use. Additionally, you can override the tool description (exposed by the MCP server) by specifying the description parameter.

Tip

Use the _validate_tool_exist_on_server parameter to validate whether the tool is available or not at instantiation time.

Running the Flow

Execute the flow as follows:

inputs = {}
conversation = assistant.start_conversation(inputs=inputs)

status = conversation.execute()
if isinstance(status, FinishedStatus):
    flow_outputs = status.output_values
    print(f"---\nFlow outputs >>> {flow_outputs}\n---")
else:
    print(
        f"Invalid execution status, expected FinishedStatus, received {type(status)}"
    )

Exporting/Loading with Agent Spec

You can export the flow configuration to Agent Spec YAML:

from wayflowcore.agentspec import AgentSpecExporter

serialized_assistant = AgentSpecExporter().to_json(assistant)

Here is what the Agent Spec representation will look like ↓

Click here to see the assistant configuration.

JSONYAML

{
  "component_type": "Flow",
  "id": "c535a5db-e2bc-4f0b-896d-d4266237428d",
  "name": "flow_34eae46e__auto",
  "description": "",
  "metadata": {
    "__metadata_info__": {}
  },
  "inputs": [],
  "outputs": [
    {
      "type": "string",
      "title": "tool_output"
    }
  ],
  "start_node": {
    "$component_ref": "d8948188-ea98-4bdb-aaa4-9c29f600696b"
  },
  "nodes": [
    {
      "$component_ref": "50108849-6be7-48d8-b3d0-5bf78c12403c"
    },
    {
      "$component_ref": "d8948188-ea98-4bdb-aaa4-9c29f600696b"
    },
    {
      "$component_ref": "d84bc934-90bd-421a-8ce5-f7224008d0b5"
    }
  ],
  "control_flow_connections": [
    {
      "component_type": "ControlFlowEdge",
      "id": "36f9ecb6-04f8-4351-81c6-1aab04963b0a",
      "name": "__StartStep___to_mcp_tool_step_control_flow_edge",
      "description": null,
      "metadata": {
        "__metadata_info__": {}
      },
      "from_node": {
        "$component_ref": "d8948188-ea98-4bdb-aaa4-9c29f600696b"
      },
      "from_branch": null,
      "to_node": {
        "$component_ref": "50108849-6be7-48d8-b3d0-5bf78c12403c"
      }
    },
    {
      "component_type": "ControlFlowEdge",
      "id": "07883a38-48b0-43ad-aeeb-1a937786a1ff",
      "name": "mcp_tool_step_to_None End node_control_flow_edge",
      "description": null,
      "metadata": {},
      "from_node": {
        "$component_ref": "50108849-6be7-48d8-b3d0-5bf78c12403c"
      },
      "from_branch": null,
      "to_node": {
        "$component_ref": "d84bc934-90bd-421a-8ce5-f7224008d0b5"
      }
    }
  ],
  "data_flow_connections": [
    {
      "component_type": "DataFlowEdge",
      "id": "e4820094-c97f-4805-a0ac-8048a2fcaaeb",
      "name": "mcp_tool_step_tool_output_to_None End node_tool_output_data_flow_edge",
      "description": null,
      "metadata": {},
      "source_node": {
        "$component_ref": "50108849-6be7-48d8-b3d0-5bf78c12403c"
      },
      "source_output": "tool_output",
      "destination_node": {
        "$component_ref": "d84bc934-90bd-421a-8ce5-f7224008d0b5"
      },
      "destination_input": "tool_output"
    }
  ],
  "$referenced_components": {
    "d84bc934-90bd-421a-8ce5-f7224008d0b5": {
      "component_type": "EndNode",
      "id": "d84bc934-90bd-421a-8ce5-f7224008d0b5",
      "name": "None End node",
      "description": "End node representing all transitions to None in the WayFlow flow",
      "metadata": {},
      "inputs": [
        {
          "type": "string",
          "title": "tool_output"
        }
      ],
      "outputs": [
        {
          "type": "string",
          "title": "tool_output"
        }
      ],
      "branches": [],
      "branch_name": "next"
    },
    "50108849-6be7-48d8-b3d0-5bf78c12403c": {
      "component_type": "ExtendedToolNode",
      "id": "50108849-6be7-48d8-b3d0-5bf78c12403c",
      "name": "mcp_tool_step",
      "description": "",
      "metadata": {
        "__metadata_info__": {}
      },
      "inputs": [],
      "outputs": [
        {
          "type": "string",
          "title": "tool_output"
        }
      ],
      "branches": [
        "next"
      ],
      "tool": {
        "component_type": "PluginMCPTool",
        "id": "d7346aa2-3276-49f7-81a2-7b1fe13dbb7d",
        "name": "generate_random_string",
        "description": "Tool to return a random string",
        "metadata": {
          "__metadata_info__": {}
        },
        "inputs": [],
        "outputs": [
          {
            "type": "string",
            "title": "tool_output"
          }
        ],
        "client_transport": {
          "component_type": "PluginSSETransport",
          "id": "0ba05bef-c297-4bd8-ac37-738838218cf4",
          "name": "mcp_client_transport",
          "description": null,
          "metadata": {},
          "url": "http://localhost:8001/sse",
          "headers": null,
          "timeout": 5.0,
          "sse_read_timeout": 300.0,
          "component_plugin_name": "MCPPlugin",
          "component_plugin_version": "25.4.0.dev0"
        },
        "component_plugin_name": "MCPPlugin",
        "component_plugin_version": "25.4.0.dev0"
      },
      "input_mapping": {},
      "output_mapping": {},
      "raise_exceptions": false,
      "component_plugin_name": "NodesPlugin",
      "component_plugin_version": "25.4.0.dev0"
    },
    "d8948188-ea98-4bdb-aaa4-9c29f600696b": {
      "component_type": "StartNode",
      "id": "d8948188-ea98-4bdb-aaa4-9c29f600696b",
      "name": "__StartStep__",
      "description": "",
      "metadata": {
        "__metadata_info__": {}
      },
      "inputs": [],
      "outputs": [],
      "branches": [
        "next"
      ]
    }
  },
  "agentspec_version": "25.4.1"
}

You can then load the configuration back to an assistant using the AgentSpecLoader.

from wayflowcore.agentspec import AgentSpecLoader

assistant: Flow = AgentSpecLoader().load_json(serialized_assistant)

Note

This guide uses the following extension/plugin Agent Spec components:

• PluginMCPTool

• PluginSSETransport

• ExtendedToolNode

See the list of available Agent Spec extension/plugin components in the API Reference

Next Steps

Having learned how to integrate MCP servers in WayFlow, you may now proceed to:

• Agents in Flows

• How to Add User Confirmation to Tool Call Requests

• How to Create a ServerTool from a Flow

Full code

Click on the card at the top of this page to download the full code for this guide or copy the code below.

# Copyright © 2025 Oracle and/or its affiliates.
#
# This software is under the Universal Permissive License
# %%[markdown]
# WayFlow Code Example - How to connect MCP tools to Assistants
# -------------------------------------------------------------

# How to use:
# Create a new Python virtual environment and install the latest WayFlow version.
# ```bash
# python -m venv venv-wayflowcore
# source venv-wayflowcore/bin/activate
# pip install --upgrade pip
# pip install "wayflowcore==26.1" 
# ```

# You can now run the script
# 1. As a Python file:
# ```bash
# python howto_mcp.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# (UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl) or Apache License
# 2.0 (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0), at your option.




# %%[markdown]
##Create a MCP Server

# %%
from mcp.server.fastmcp import FastMCP

PAYSLIPS = [
    {
        "Amount": 7612,
        "Currency": "USD",
        "PeriodStartDate": "2025/05/15",
        "PeriodEndDate": "2025/06/15",
        "PaymentDate": "",
        "DocumentId": 2,
        "PersonId": 2,
    },
    {
        "Amount": 5000,
        "Currency": "CHF",
        "PeriodStartDate": "2024/05/01",
        "PeriodEndDate": "2024/06/01",
        "PaymentDate": "2024/05/15",
        "DocumentId": 1,
        "PersonId": 1,
    },
    {
        "Amount": 10000,
        "Currency": "EUR",
        "PeriodStartDate": "2025/06/15",
        "PeriodEndDate": "2025/10/15",
        "PaymentDate": "",
        "DocumentsId": 3,
        "PersonId": 3,
    },
]

def create_server(host: str, port: int):
    """Create and configure the MCP server"""
    server = FastMCP(
        name="Example MCP Server",
        instructions="A MCP Server.",
        host=host,
        port=port,
    )

    @server.tool(description="Return session details for the current user")
    def get_user_session():
        return {
            "PersonId": "1",
            "Username": "Bob.b",
            "DisplayName": "Bob B",
        }

    @server.tool(description="Return payslip details for a given PersonId")
    def get_payslips(PersonId: int):
        return [payslip for payslip in PAYSLIPS if payslip["PersonId"] == int(PersonId)]

    return server


def start_mcp_server() -> str:
    host: str = "localhost"
    port: int = 8080
    server = create_server(host=host, port=port)
    server.run(transport="sse")

    return f"http://{host}:{port}/sse"

# mcp_server_url = start_mcp_server() # <--- Move the code above to a separate file then uncomment

# %%[markdown]
## Imports for this guide

# %%
from wayflowcore.executors.executionstatus import FinishedStatus, UserMessageRequestStatus
from wayflowcore.agent import Agent
from wayflowcore.mcp import MCPTool, MCPToolBox, SSETransport, enable_mcp_without_auth
from wayflowcore.flow import Flow
from wayflowcore.steps import ToolExecutionStep

mcp_server_url = f"http://localhost:8080/sse" # change to your own URL
# We will see below how to connect a specific tool to an assistant, e.g.
MCP_TOOL_NAME = "get_user_session"
# And see how to build an agent that can answer questions, e.g.
USER_QUERY = "What was the payment date of the last payslip for the current user?"

# %%[markdown]
## Configure your LLM

# %%
from wayflowcore.models import VllmModel
llm = VllmModel(
    model_id="LLAMA_MODEL_ID",
    host_port="LLAMA_API_URL",
)


# %%[markdown]
## Connecting an agent to the MCP server

# %%
enable_mcp_without_auth() # <--- See https://modelcontextprotocol.io/specification/2025-06-18/basic/authorization#security-considerations
mcp_client = SSETransport(url=mcp_server_url)
mcp_toolbox = MCPToolBox(client_transport=mcp_client)

assistant = Agent(
    llm=llm,
    tools=[mcp_toolbox]
)


# %%[markdown]
## Running the agent

# %%
# With a linear conversation
conversation = assistant.start_conversation()

conversation.append_user_message(USER_QUERY)
status = conversation.execute()
if isinstance(status, UserMessageRequestStatus):
    assistant_reply = conversation.get_last_message()
    print(f"---\nAssistant >>> {assistant_reply.content}\n---")
else:
    print(f"Invalid execution status, expected UserMessageRequestStatus, received {type(status)}")

# then continue the conversation

# %%[markdown]
## Running with an execution loop

# %%
def run_agent_in_command_line(assistant: Agent):
    inputs = {}
    conversation = assistant.start_conversation(inputs)

    while True:
        status = conversation.execute()
        if isinstance(status, FinishedStatus):
            break
        assistant_reply = conversation.get_last_message()
        if assistant_reply is not None:
            print("\nAssistant >>>", assistant_reply.content)
        user_input = input("\nUser >>> ")
        conversation.append_user_message(user_input)

# run_agent_in_command_line(assistant)
# ^ uncomment and execute

# %%[markdown]
## Connecting a flow to the MCP server

# %%
mcp_tool = MCPTool(
    name=MCP_TOOL_NAME,
    client_transport=mcp_client
)

assistant = Flow.from_steps([
    ToolExecutionStep(name="mcp_tool_step", tool=mcp_tool)
])

# %%[markdown]
## Running the flow

# %%
inputs = {}
conversation = assistant.start_conversation(inputs=inputs)

status = conversation.execute()
if isinstance(status, FinishedStatus):
    flow_outputs = status.output_values
    print(f"---\nFlow outputs >>> {flow_outputs}\n---")
else:
    print(
        f"Invalid execution status, expected FinishedStatus, received {type(status)}"
    )

# %%[markdown]
## Export config to Agent Spec

# %%
from wayflowcore.agentspec import AgentSpecExporter

serialized_assistant = AgentSpecExporter().to_json(assistant)

# %%[markdown]
## Load Agent Spec config

# %%
from wayflowcore.agentspec import AgentSpecLoader

assistant: Flow = AgentSpecLoader().load_json(serialized_assistant)