How to connect MCP tools to Assistants#

python-icon Download Python Script

Python script/notebook for this guide.

MCP how-to script

Prerequisites

This guide assumes familiarity with:

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: Set up 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, authless_mcp_enabled
from wayflowcore.flow import Flow
from wayflowcore.retrypolicy import RetryPolicy
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"
MCP_TOOLBOX_TOOL_FILTER = ["get_user_session", "get_payslips"]
# 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:

from wayflowcore.models import OCIGenAIModel, OCIClientConfigWithApiKey

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

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).

mcp_client = SSETransport(url=mcp_server_url)
with authless_mcp_enabled(): # <--- See https://modelcontextprotocol.io/specification/2025-06-18/basic/authorization#security-considerations
    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.

If the MCP server composes tools from external MCP servers and can temporarily return a partial tool list during health or remount events, declare the expected tools with tool_filter and configure retry_policy. When an expected tool is missing from a successful list_tools response, WayFlow retries the tool-list resolution before failing. The same policy is propagated to generated MCPTool instances for transient tool execution failures.

mcp_toolbox_with_retries = MCPToolBox(
    client_transport=mcp_client,
    tool_filter=MCP_TOOLBOX_TOOL_FILTER,
    retry_policy=RetryPolicy(
        max_attempts=3,
        initial_retry_delay=0.25,
        max_retry_delay=2.0,
    ),
)

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

Note

Tool-list missing-tool retry only applies when WayFlow knows which tools are expected, for example through tool_filter or a direct MCPTool(name=...). If tool_filter is None, WayFlow cannot determine whether a successful tool-list response is incomplete.

Note

The MCP tool retry policy is separate from transport-level retry. It handles successful tool-list responses that are missing expected tools and transient tool execution failures. Configure retry_policy on the transport when you need lower-level HTTP client retry or request timeout behavior.

Note

authless_mcp_enabled() disables authorization for local/testing only—do not use in production. Keep it scoped around the code that creates MCP tools or toolboxes.

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

Note

WayFlow maintains MCP Client sessions between calls, which means that the client does not need to re-authenticate at every call. After establishing a secure connection, MCP servers can safely perform session recognition (e.g. for retrieving user information)

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:

with authless_mcp_enabled():
    mcp_tool = MCPTool(
        name=MCP_TOOL_NAME,
        client_transport=mcp_client
    )
# .. start-##_Configuring_direct_tool_retry_policy
with authless_mcp_enabled():
    mcp_tool_with_retries = MCPTool(
        name=MCP_TOOL_NAME,
        client_transport=mcp_client,
        retry_policy=RetryPolicy(max_attempts=3),
    )
# .. end-##_Configuring_direct_tool_retry_policy

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

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. You can also pass retry_policy to retry direct tool resolution and transient tool execution failures.

with authless_mcp_enabled():
    mcp_tool_with_retries = MCPTool(
        name=MCP_TOOL_NAME,
        client_transport=mcp_client,
        retry_policy=RetryPolicy(max_attempts=3),
    )

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)}"
    )

Advanced use: Use OAuth in MCP Tools#

MCP Tools and ToolBoxes support auth using the official OAuth flow from MCP,

To enable auth simply provide an Auth configuration to an MCP Client Transport (SSE or StreamableHTTP).

import webbrowser
from wayflowcore.mcp import MCPOAuthConfigFactory

oauth_callback_port = 8001 # depends on your MCP server configuration
auth = MCPOAuthConfigFactory.with_dynamic_discovery(
    redirect_uri=f"http://localhost:{oauth_callback_port}/callback"
)
client_transport = SSETransport(url=sse_mcp_server_oauth, auth=auth)

tool = MCPTool(
    name="generate_random_string",
    description="1234567",
    client_transport=client_transport,
    input_descriptors=[],
    # ^ make sure to specify the input/output descriptors for the tool
)

agent = Agent(llm=llm, tools=[tool])

Important

You must disable MCPTool verification at instantiation when using OAuth.

Then when running the assistant, when authorization is required an execution status is returned with the authorization URL. The client is responsible for obtaining the auth code and state and submit it back to the execution loop, which will complete the OAuth flow. Once the OAuth flow is completed, the conversation can be resumed with the now authenticated MCP Client sessions.

from wayflowcore.auth.auth import AuthChallengeResult
from wayflowcore.executors.executionstatus import AuthChallengeRequestStatus

conv = agent.start_conversation()
conv.append_user_message("Call the tool please")
status = conv.execute()

assert isinstance(status, AuthChallengeRequestStatus)
authorization_url = status.auth_request.authorization_url

# The client app must consume the authorization URL, fetch the code/state
# and submit it back to complete the OAuth flow.
webbrowser.open(authorization_url)
auth_code, auth_state = ...

# The auth challenge result is submitted, which completes the auth flow
status.submit_result(AuthChallengeResult(code=auth_code, state=auth_state))

# The conversation is resumed, and returns the expected result
status = conv.execute()

API Reference: OAuthClientConfig

OAuth works with MCP Tools and ToolBoxes, in agents, flows and multi-agent patterns.

Note

Note that MCP client sessions are reused in a single conversation, which means that you will not have to re-perform the OAuth flow at every request.

Advanced use: Complex types in MCP tools#

WayFlow supports MCP tools with non-string outputs, such as:

  • List of string

  • Dictionary with keys and values of string type

From the MCP server-side, you may need to enable the structured_output parameter of your MCP server (depending on the implementation).

server = FastMCP(
    name="Example MCP Server",
    instructions="A MCP Server.",
    host=host,
    port=port,
)

@server.tool(description="Tool that generates a dictionary", structured_output=True)
def generate_dict() -> dict[str, str]:
    return {"key": "value"}

@server.tool(description="Tool that generates a list", structured_output=True)
def generate_list() -> list[str]:
    return ["value1", "value2"]

On the WayFlow side, the input and output descriptors can be automatically inferred.

generate_dict_tool = MCPTool(
    name="generate_dict",
    description="Tool that generates a dictionary",
    client_transport=mcp_client,
    # output_descriptors=[DictProperty(name="generate_dictOutput")], # this will be automatically inferred
)

generate_list_tool = MCPTool(
    name="generate_list",
    description="Tool that generates a list",
    client_transport=mcp_client,
    # output_descriptors=[ListProperty(name="generate_listOutput")], # this will be automatically inferred
)

You can then use those tools in a Flow to natively support the manipulation of complex data types with MCP tools.

You can also use Pydantic models to change the tool output names. Note that in this advanced use, you must wrap the outputs in a result field as expected by MCP when using non-dict types. This also enables the use of multi-output in tools by using tuples.

from typing import Annotated
from pydantic import BaseModel, RootModel, Field

class GenerateTupleOut(BaseModel, title="tool_output"):
    result: tuple[
        Annotated[str, Field(title="str_output")],
        Annotated[bool, Field(title="bool_output")]
    ]
    # /!\ this needs to be named `result`

class GenerateListOut(BaseModel, title="tool_output"):
    result: list[str] # /!\ this needs to be named `result`

class GenerateDictOut(RootModel[dict[str, str]], title="tool_output"):
    pass

server = FastMCP(
    name="Example MCP Server",
    instructions="A MCP Server.",
    host=host,
    port=port,
)

@server.tool(description="Tool that generates a dictionary", structured_output=True)
def generate_dict() -> GenerateDictOut:
    return GenerateDictOut({"key": "value"})

@server.tool(description="Tool that generates a list", structured_output=True)
def generate_list() -> GenerateListOut:
    return GenerateListOut(result=["value1", "value2"])

@server.tool(description="Tool that returns multiple outputs", structured_output=True)
def generate_tuple(inputs: list[str]) -> GenerateTupleOut:
    value = "; ".join(inputs)
    return GenerateTupleOut(result=("value", True))

You can then match the output descriptors on the WayFlow side.

generate_dict_tool = MCPTool(
    name="generate_dict",
    description="Tool that generates a dictionary",
    client_transport=mcp_client,
    output_descriptors=[DictProperty(name="tool_output")],
)

generate_list_tool = MCPTool(
    name="generate_list",
    description="Tool that generates a list",
    client_transport=mcp_client,
    output_descriptors=[ListProperty(name="tool_output")],
)

generate_tuple_tool = MCPTool(
    name="generate_tuple",
    description="Tool that returns multiple outputs",
    client_transport=mcp_client,
    input_descriptors=[ListProperty(name="inputs")],
    output_descriptors=[StringProperty(name="str_output"), BooleanProperty(name="bool_output")],
)

When specified, the input/output descriptors of the MCP tool will be validated against the schema fetched from the MCP server.

Note

MCPToolBox is not compatible with complex output types. Tools from MCPToolBox will always return string values.

Exporting/Loading with Agent Spec#

You can export the assistant from this tutorial to Agent Spec:

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.
{
    "component_type": "ExtendedAgent",
    "id": "024a2c39-3695-450f-a950-eabc1663db17",
    "name": "agent_b2a01d24__auto",
    "description": "",
    "metadata": {
        "__metadata_info__": {}
    },
    "inputs": [],
    "outputs": [],
    "llm_config": {
        "component_type": "VllmConfig",
        "id": "f42c7884-3cfb-4f82-86eb-1cea1a5a51c6",
        "name": "LLAMA_MODEL_ID",
        "description": null,
        "metadata": {
            "__metadata_info__": {}
        },
        "default_generation_parameters": {
            "max_tokens": 512
        },
        "url": "LLAMA_API_URL",
        "model_id": "LLAMA_MODEL_ID"
    },
    "system_prompt": "",
    "tools": [],
    "toolboxes": [
        {
            "component_type": "PluginMCPToolBox",
            "id": "43a0c0dc-2638-4dee-b868-44dd2b9afa35",
            "name": "",
            "description": null,
            "metadata": {},
            "client_transport": {
                "component_type": "SSETransport",
                "id": "ca2e9a97-6e1a-4f11-931a-774826c975ad",
                "name": "mcp_client_transport",
                "description": null,
                "metadata": {},
                "session_parameters": {
                    "read_timeout_seconds": 60
                },
                "url": "http://localhost:61799/sse",
                "headers": null
            },
            "tool_filter": null,
            "component_plugin_name": "MCPPlugin",
            "component_plugin_version": "25.4.1"
        }
    ],
    "context_providers": null,
    "can_finish_conversation": false,
    "max_iterations": 10,
    "initial_message": "Hi! How can I help you?",
    "caller_input_mode": "always",
    "agents": [],
    "flows": [],
    "agent_template": {
        "component_type": "PluginPromptTemplate",
        "id": "b33b40c4-02d2-4f91-b60a-d3ea1c4d725d",
        "name": "",
        "description": null,
        "metadata": {
            "__metadata_info__": {}
        },
        "messages": [
            {
                "role": "system",
                "contents": [
                    {
                        "type": "text",
                        "content": "{%- if __TOOLS__ -%}\nEnvironment: ipython\nCutting Knowledge Date: December 2023\n\nYou are a helpful assistant with tool calling capabilities. Only reply with a tool call if the function exists in the library provided by the user. If it doesn't exist, just reply directly in natural language. When you receive a tool call response, use the output to format an answer to the original user question.\n\nYou have access to the following functions. To call a function, please respond with JSON for a function call.\nRespond in the format {\"name\": function name, \"parameters\": dictionary of argument name and its value}.\nDo not use variables.\n\n[{% for tool in __TOOLS__%}{{tool.to_openai_format() | tojson}}{{', ' if not loop.last}}{% endfor %}]\n{%- endif -%}\n"
                    }
                ],
                "tool_requests": null,
                "tool_result": null,
                "display_only": false,
                "sender": null,
                "recipients": [],
                "time_created": "2025-10-07T08:08:34.942842+00:00",
                "time_updated": "2025-10-07T08:08:34.942844+00:00"
            },
            {
                "role": "system",
                "contents": [
                    {
                        "type": "text",
                        "content": "{%- if custom_instruction -%}Additional instructions:\n{{custom_instruction}}{%- endif -%}"
                    }
                ],
                "tool_requests": null,
                "tool_result": null,
                "display_only": false,
                "sender": null,
                "recipients": [],
                "time_created": "2025-10-07T08:08:34.942869+00:00",
                "time_updated": "2025-10-07T08:08:34.942870+00:00"
            },
            {
                "role": "system",
                "contents": [
                    {
                        "type": "text",
                        "content": "$$__CHAT_HISTORY_PLACEHOLDER__$$"
                    }
                ],
                "tool_requests": null,
                "tool_result": null,
                "display_only": false,
                "sender": null,
                "recipients": [],
                "time_created": "2025-10-07T08:08:34.934280+00:00",
                "time_updated": "2025-10-07T08:08:34.934523+00:00"
            },
            {
                "role": "system",
                "contents": [
                    {
                        "type": "text",
                        "content": "{% if __PLAN__ %}The current plan you should follow is the following: \n{{__PLAN__}}{% endif %}"
                    }
                ],
                "tool_requests": null,
                "tool_result": null,
                "display_only": false,
                "sender": null,
                "recipients": [],
                "time_created": "2025-10-07T08:08:34.942886+00:00",
                "time_updated": "2025-10-07T08:08:34.942886+00:00"
            }
        ],
        "output_parser": {
            "component_type": "PluginJsonToolOutputParser",
            "id": "b5e2e684-7434-4efd-857d-06d9172c2bab",
            "name": "jsontool_outputparser",
            "description": null,
            "metadata": {
                "__metadata_info__": {}
            },
            "tools": null,
            "component_plugin_name": "OutputParserPlugin",
            "component_plugin_version": "25.4.1"
        },
        "inputs": [
            {
                "description": "\"__TOOLS__\" input variable for the template",
                "title": "__TOOLS__"
            },
            {
                "description": "\"custom_instruction\" input variable for the template",
                "type": "string",
                "title": "custom_instruction"
            },
            {
                "description": "\"__PLAN__\" input variable for the template",
                "type": "string",
                "title": "__PLAN__",
                "default": ""
            },
            {
                "type": "array",
                "items": {},
                "title": "__CHAT_HISTORY__"
            }
        ],
        "pre_rendering_transforms": null,
        "post_rendering_transforms": [
            {
                "component_type": "PluginRemoveEmptyNonUserMessageTransform",
                "id": "10c7ba9e-49a7-4c1e-b317-081daa14e771",
                "name": "removeemptynonusermessage_messagetransform",
                "description": null,
                "metadata": {
                    "__metadata_info__": {}
                },
                "component_plugin_name": "MessageTransformPlugin",
                "component_plugin_version": "25.4.1"
            },
            {
                "component_type": "PluginCoalesceSystemMessagesTransform",
                "id": "f3ac574f-8794-4a2c-8087-3ed21894b4bf",
                "name": "coalescesystemmessage_messagetransform",
                "description": null,
                "metadata": {
                    "__metadata_info__": {}
                },
                "component_plugin_name": "MessageTransformPlugin",
                "component_plugin_version": "25.4.1"
            },
            {
                "component_type": "PluginLlamaMergeToolRequestAndCallsTransform",
                "id": "baf2fd46-ca48-4a73-a8af-98cf805ecb08",
                "name": "llamamergetoolrequestandcalls_messagetransform",
                "description": null,
                "metadata": {
                    "__metadata_info__": {}
                },
                "component_plugin_name": "MessageTransformPlugin",
                "component_plugin_version": "25.4.1"
            }
        ],
        "tools": null,
        "native_tool_calling": false,
        "response_format": null,
        "native_structured_generation": true,
        "generation_config": null,
        "component_plugin_name": "PromptTemplatePlugin",
        "component_plugin_version": "25.4.1"
    },
    "component_plugin_name": "AgentPlugin",
    "component_plugin_version": "25.4.1",
    "agentspec_version": "25.4.1"
}

You can then load the configuration back to an assistant using the AgentSpecLoader. Because this example uses an unauthenticated local MCP server, the load step is wrapped in authless_mcp_enabled() too.

from wayflowcore.agentspec import AgentSpecLoader

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

Note

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

  • PluginMCPToolBox

  • 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:

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.

  1# Copyright © 2025, 2026 Oracle and/or its affiliates.
  2#
  3# This software is under the Apache License 2.0
  4# %%[markdown]
  5# WayFlow Code Example - How to connect MCP tools to Assistants
  6# -------------------------------------------------------------
  7
  8# How to use:
  9# Create a new Python virtual environment and install the latest WayFlow version.
 10# ```bash
 11# python -m venv venv-wayflowcore
 12# source venv-wayflowcore/bin/activate
 13# pip install --upgrade pip
 14# pip install "wayflowcore==26.3.0" 
 15# ```
 16
 17# You can now run the script
 18# 1. As a Python file:
 19# ```bash
 20# python howto_mcp.py
 21# ```
 22# 2. As a Notebook (in VSCode):
 23# When viewing the file,
 24#  - press the keys Ctrl + Enter to run the selected cell
 25#  - or Shift + Enter to run the selected cell and move to the cell below# (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License
 26# (UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option.
 27
 28
 29
 30
 31# %%[markdown]
 32##Create a MCP Server
 33
 34# %%
 35from mcp.server.fastmcp import FastMCP
 36
 37PAYSLIPS = [
 38    {
 39        "Amount": 7612,
 40        "Currency": "USD",
 41        "PeriodStartDate": "2025/05/15",
 42        "PeriodEndDate": "2025/06/15",
 43        "PaymentDate": "",
 44        "DocumentId": 2,
 45        "PersonId": 2,
 46    },
 47    {
 48        "Amount": 5000,
 49        "Currency": "CHF",
 50        "PeriodStartDate": "2024/05/01",
 51        "PeriodEndDate": "2024/06/01",
 52        "PaymentDate": "2024/05/15",
 53        "DocumentId": 1,
 54        "PersonId": 1,
 55    },
 56    {
 57        "Amount": 10000,
 58        "Currency": "EUR",
 59        "PeriodStartDate": "2025/06/15",
 60        "PeriodEndDate": "2025/10/15",
 61        "PaymentDate": "",
 62        "DocumentsId": 3,
 63        "PersonId": 3,
 64    },
 65]
 66
 67def create_server(host: str, port: int):
 68    """Create and configure the MCP server"""
 69    server = FastMCP(
 70        name="Example MCP Server",
 71        instructions="A MCP Server.",
 72        host=host,
 73        port=port,
 74    )
 75
 76    @server.tool(description="Return session details for the current user")
 77    def get_user_session():
 78        return {
 79            "PersonId": "1",
 80            "Username": "Bob.b",
 81            "DisplayName": "Bob B",
 82        }
 83
 84    @server.tool(description="Return payslip details for a given PersonId")
 85    def get_payslips(PersonId: int):
 86        return [payslip for payslip in PAYSLIPS if payslip["PersonId"] == int(PersonId)]
 87
 88    return server
 89
 90
 91def start_mcp_server() -> str:
 92    host: str = "localhost"
 93    port: int = 8080
 94    server = create_server(host=host, port=port)
 95    server.run(transport="sse")
 96
 97    return f"http://{host}:{port}/sse"
 98
 99# mcp_server_url = start_mcp_server() # <--- Move the code above to a separate file then uncomment
100
101# %%[markdown]
102## Imports for this guide
103
104# %%
105from wayflowcore.executors.executionstatus import FinishedStatus, UserMessageRequestStatus
106from wayflowcore.agent import Agent
107from wayflowcore.mcp import MCPTool, MCPToolBox, SSETransport, authless_mcp_enabled
108from wayflowcore.flow import Flow
109from wayflowcore.retrypolicy import RetryPolicy
110from wayflowcore.steps import ToolExecutionStep
111
112mcp_server_url = f"http://localhost:8080/sse" # change to your own URL
113# We will see below how to connect a specific tool to an assistant, e.g.
114MCP_TOOL_NAME = "get_user_session"
115MCP_TOOLBOX_TOOL_FILTER = ["get_user_session", "get_payslips"]
116# And see how to build an agent that can answer questions, e.g.
117USER_QUERY = "What was the payment date of the last payslip for the current user?"
118
119# %%[markdown]
120## Configure your LLM
121
122# %%
123from wayflowcore.models import VllmModel
124llm = VllmModel(
125    model_id="LLAMA_MODEL_ID",
126    host_port="LLAMA_API_URL",
127)
128
129
130# %%[markdown]
131## Connecting an agent to the MCP server
132
133# %%
134mcp_client = SSETransport(url=mcp_server_url)
135with authless_mcp_enabled(): # <--- See https://modelcontextprotocol.io/specification/2025-06-18/basic/authorization#security-considerations
136    mcp_toolbox = MCPToolBox(client_transport=mcp_client)
137
138assistant = Agent(
139    llm=llm,
140    tools=[mcp_toolbox]
141)
142
143# %%[markdown]
144## Configuring toolbox retry policy
145
146# %%
147mcp_toolbox_with_retries = MCPToolBox(
148    client_transport=mcp_client,
149    tool_filter=MCP_TOOLBOX_TOOL_FILTER,
150    retry_policy=RetryPolicy(
151        max_attempts=3,
152        initial_retry_delay=0.25,
153        max_retry_delay=2.0,
154    ),
155)
156
157assistant = Agent(
158    llm=llm,
159    tools=[mcp_toolbox_with_retries]
160)
161
162
163# %%[markdown]
164## Running the agent
165
166# %%
167# With a linear conversation
168conversation = assistant.start_conversation()
169
170conversation.append_user_message(USER_QUERY)
171status = conversation.execute()
172if isinstance(status, UserMessageRequestStatus):
173    assistant_reply = conversation.get_last_message()
174    print(f"---\nAssistant >>> {assistant_reply.content}\n---")
175else:
176    print(f"Invalid execution status, expected UserMessageRequestStatus, received {type(status)}")
177
178# then continue the conversation
179
180# %%[markdown]
181## Running with an execution loop
182
183# %%
184def run_agent_in_command_line(assistant: Agent):
185    inputs = {}
186    conversation = assistant.start_conversation(inputs)
187
188    while True:
189        status = conversation.execute()
190        if isinstance(status, FinishedStatus):
191            break
192        assistant_reply = conversation.get_last_message()
193        if assistant_reply is not None:
194            print("\nAssistant >>>", assistant_reply.content)
195        user_input = input("\nUser >>> ")
196        conversation.append_user_message(user_input)
197
198# run_agent_in_command_line(assistant)
199# ^ uncomment and execute
200
201# %%[markdown]
202## Connecting a flow to the MCP server
203
204# %%
205with authless_mcp_enabled():
206    mcp_tool = MCPTool(
207        name=MCP_TOOL_NAME,
208        client_transport=mcp_client
209    )
210
211# %%[markdown]
212## Configuring direct tool retry policy
213
214# %%
215with authless_mcp_enabled():
216    mcp_tool_with_retries = MCPTool(
217        name=MCP_TOOL_NAME,
218        client_transport=mcp_client,
219        retry_policy=RetryPolicy(max_attempts=3),
220    )
221
222assistant = Flow.from_steps([
223    ToolExecutionStep(name="mcp_tool_step", tool=mcp_tool_with_retries)
224])
225# .. end-##_Connecting_a_flow_to_the_MCP_server
226
227# %%[markdown]
228## Running the flow
229
230# %%
231inputs = {}
232conversation = assistant.start_conversation(inputs=inputs)
233
234status = conversation.execute()
235if isinstance(status, FinishedStatus):
236    flow_outputs = status.output_values
237    print(f"---\nFlow outputs >>> {flow_outputs}\n---")
238else:
239    print(
240        f"Invalid execution status, expected FinishedStatus, received {type(status)}"
241    )
242
243# %%[markdown]
244## Export config to Agent Spec
245
246# %%
247from wayflowcore.agentspec import AgentSpecExporter
248
249serialized_assistant = AgentSpecExporter().to_json(assistant)
250
251# %%[markdown]
252## Load Agent Spec config
253
254# %%
255from wayflowcore.agentspec import AgentSpecLoader
256
257with authless_mcp_enabled():
258    assistant: Flow = AgentSpecLoader().load_json(serialized_assistant)