WayFlow

https://oracle.github.io/wayflow/
Admin
WayFlow is a powerful Python library for building advanced AI-powered assistants using modular...
Tokens:426,233
Snippets:3,610
Trust Score:9.7
Update:5 months ago
Show doc for...
Tokens:
Raw
### Initialize and Use Agent

Source: https://oracle.github.io/wayflow/26.1.1/core/api/agent.html

Demonstrates how to initialize an Agent with an LLM, start a conversation, add a user message, and execute the conversation to get an agent's response. Ensure the LLM is properly configured before initialization.

```python
>>> from wayflowcore.agent import Agent
>>> agent = Agent(llm=llm)
>>> conversation = agent.start_conversation()
>>> conversation.append_user_message("I need help regarding my sql query")
>>> status = conversation.execute()
>>> agent_answer = conversation.get_last_message().content
>>> # I'd be happy to help with your SQL query
```

--------------------------------

### ToolContextProvider Example

Source: https://oracle.github.io/wayflow/26.1.1/core/api/contextproviders.html

Demonstrates how to wrap a tool execution using ToolContextProvider. This setup includes defining a tool, creating the context provider, and integrating it into a Flow with output steps to display the tool's result.

```python
>>> from time import time
>>> from wayflowcore.controlconnection import ControlFlowEdge
>>> from wayflowcore.contextproviders import ToolContextProvider
>>> from wayflowcore.flow import Flow
>>> from wayflowcore.steps import OutputMessageStep
>>> from wayflowcore.tools import tool
>>>
>>> @tool(description_mode='only_docstring')
... def current_time() -> str:
...     '''Tool that returns time'''
...     from time import time
...     return str(time())
>>>
>>> context_provider = ToolContextProvider(tool=current_time, output_name="time_output_io", name="tool_step")
>>> display_first_step_time = OutputMessageStep(message_template="{{ time_output_io }}", name="display_first_step_time")
>>>
>>> display_second_step_time = OutputMessageStep(
...     message_template="{{ time_output_io }}", name="display_second_step_time"
... )
>>> flow = Flow(
...     begin_step=display_first_step_time,
...     control_flow_edges=[
...         ControlFlowEdge(
...             source_step=display_first_step_time,
...             destination_step=display_second_step_time,
...         ),
...         ControlFlowEdge(source_step=display_second_step_time, destination_step=None),
...     ],
...     context_providers=[context_provider],
... )
```

--------------------------------

### DataFlowEdge Example Initialization

Source: https://oracle.github.io/wayflow/26.1.1/core/api/flows.html

Example demonstrating the import and instantiation of DataFlowEdge along with related Wayflow core classes. This setup is typical for defining data connections within a flow.

```python
>>> from wayflowcore.controlconnection import ControlFlowEdge
>>> from wayflowcore.dataconnection import DataFlowEdge
>>> from wayflowcore.flow import Flow
>>> from wayflowcore.steps import OutputMessageStep
```

--------------------------------

### Execute Basic Assistant Flow

Source: https://oracle.github.io/wayflow/26.1.1/core/tutorials/basic_flow.html

Run the assistant by starting a conversation, executing it to get the initial message, appending a user question, and executing again to get the final answer. This example assumes the assistant is configured to answer a single question and then exit.

```python
1# Start a conversation.
2conversation = assistant.start_conversation()
3
4# Execute the assistant.
5# This will print out the message to the user, then stop at the user input step.
6conversation.execute()
7
8# Ask a question of the assistant by appending a user message.
9conversation.append_user_message("Does John Smith earn more that Mary Jones?")
10
11# Execute the assistant again. Continues from the UserInputStep.
12# As there are no other steps the flow will run to the end.
13status = conversation.execute()
14
15# "output_message" is the default key name for the output value
16# of the OutputMessageStep.
17from wayflowcore.executors.executionstatus import FinishedStatus
18if isinstance(status, FinishedStatus):
19    answer = status.output_values[OutputMessageStep.OUTPUT]
20    print(answer)
21else:
22    print(
23        f"Incorrect execution status, expected FinishedStatus, got {status.__class__.__name__}"
24    )
```

--------------------------------

### Execute Flow with Different Inputs

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/catching_exceptions.html

These examples show how to start a conversation with the constructed flow and execute it with different inputs to demonstrate both successful execution and exception handling.

```python
conversation = flow.start_conversation(inputs={"raise_error": False})
conversation.execute()
# "Success: No error was raised"

conversation = flow.start_conversation(inputs={"raise_error": True})
conversation.execute()
# "Failure: Did get an error: ValueError"

```

--------------------------------

### OCI Configuration File Example

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/llm_from_different_providers.html

Example structure for the ~/.oci/config file for API key authentication. Ensure all placeholders are replaced with your specific OCI details.

```ini
[DEFAULT]
user=ocid1.user.oc1..<given in step 7>
fingerprint=<given in step 7>
tenancy=ocid1.tenancy.oc1..<given in step 7>
region=<given in step 7 (region where you created your key in step 4)>
key_file=<path of the downloaded key in step 5 (for convenience, store the key in the .oci directory and ensure it has a .pem suffix.)>
```

--------------------------------

### Define Start Step

Source: https://oracle.github.io/wayflow/26.1.1/core/tutorials/basic_flow.html

Initialize the starting point of the flow with a StartStep. This step has no input descriptors.

```python
# A start step. This is where the flow starts.
start_step = StartStep(name="start_step", input_descriptors=None)
```

--------------------------------

### FlowContextProvider Example

Source: https://oracle.github.io/wayflow/26.1.1/core/api/contextproviders.html

Illustrates wrapping a nested Flow using FlowContextProvider. This example shows how to define a contextual flow, create the provider with specified output names, and integrate it into a main Flow to access the nested flow's outputs.

```python
>>> from wayflowcore.contextproviders import FlowContextProvider
>>> from wayflowcore.flow import Flow
>>> from wayflowcore.steps import OutputMessageStep
>>> contextual_flow = Flow.from_steps([OutputMessageStep(
...     message_template="The current time is 2pm.",
...     output_mapping={OutputMessageStep.OUTPUT: "time_output_io"},
... )])
>>> context_provider = FlowContextProvider(contextual_flow, flow_output_names=["time_output_io"])
>>> flow = Flow.from_steps(
...     steps=[OutputMessageStep("Last time message: {{time_output_io}}")],
...     context_providers=[context_provider]
... )
>>> conversation = flow.start_conversation()
>>> execution_status = conversation.execute()
>>> last_message = conversation.get_last_message()
>>> # print(last_message.content)
>>> # Last time message: The current time is 2pm.
```

--------------------------------

### Initialize Datastore and Populate with Dummy Data

Source: https://oracle.github.io/wayflow/26.1.1/core/api/flows.html

Demonstrates how to set up an InMemoryDatastore with a document entity and populate it with initial data before using the DatastoreUpdateStep.

```python
from wayflowcore.datastore import Entity
from wayflowcore.datastore.inmemory import InMemoryDatastore
from wayflowcore.flowhelpers import create_single_step_flow
from wayflowcore.property import StringProperty, IntegerProperty
from wayflowcore.steps.datastoresteps import DatastoreUpdateStep

document = Entity(
    properties={ "id": IntegerProperty(), "content": StringProperty(default_value="Empty...") }
)
datastore = InMemoryDatastore({"documents": document})
dummy_data = [
    {"id": 2, "content": "The rat the cat the dog bit chased escaped."},
    {"id": 3, "content": "More people have been to Russia than I have."}
]
datastore.create("documents", dummy_data)

```

--------------------------------

### Install PyPDF Library

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_build_assistants_with_tools.html

Install the PyPDF library to enable PDF parsing functionalities. This is a prerequisite for the PDF summarizer example.

```bash
$ pip install pypdf

```

--------------------------------

### Initialize Datastore and DatastoreDeleteStep

Source: https://oracle.github.io/wayflow/26.1.1/core/api/flows.html

Demonstrates setting up an InMemoryDatastore with dummy data and then initializing a DatastoreDeleteStep to remove entities based on a 'where' clause.

```python
>>> from wayflowcore.datastore import Entity
>>> from wayflowcore.datastore.inmemory import InMemoryDatastore
>>> from wayflowcore.flowhelpers import create_single_step_flow
>>> from wayflowcore.property import StringProperty, IntegerProperty
>>> from wayflowcore.steps.datastoresteps import DatastoreDeleteStep

>>> document = Entity(
...     properties={ "id": IntegerProperty(), "content": StringProperty(default_value="Empty...") }
... )
>>> datastore = InMemoryDatastore({"documents": document})
>>> dummy_data = [
...     {"id": 2, "content": "The rat the cat the dog bit chased escaped."},
...     {"id": 3, "content": "More people have been to Russia than I have."}
... ]
>>> datastore.create("documents", dummy_data)
[{'content': 'The rat the cat the dog bit chased escaped.', 'id': 2}, {'content': 'More people have been to Russia than I have.', 'id': 3}]
```

```python
>>> datastore_delete_flow = create_single_step_flow(DatastoreDeleteStep(datastore, "documents", where={"id": 2}))
>>> conversation = datastore_delete_flow.start_conversation()
>>> execution_status = conversation.execute()
```

--------------------------------

### Initialize Datastore and Dummy Data

Source: https://oracle.github.io/wayflow/26.1.1/core/api/flows.html

Sets up an in-memory datastore with a document entity and populates it with sample data. This is a prerequisite for using DatastoreListStep.

```python
>>> from wayflowcore.datastore import Entity
>>> from wayflowcore.datastore.inmemory import InMemoryDatastore
>>> from wayflowcore.flowhelpers import create_single_step_flow
>>> from wayflowcore.property import StringProperty, IntegerProperty
>>> from wayflowcore.steps.datastoresteps import DatastoreListStep

>>> document = Entity(
...     properties={ "id": IntegerProperty(), "content": StringProperty(default_value="Empty...") }
... )
>>> datastore = InMemoryDatastore({"documents": document})
>>> dummy_data = [
...     {"id": 2, "content": "The rat the cat the dog bit chased escaped."},
...     {"id": 3, "content": "More people have been to Russia than I have."}
... ]
>>> datastore.create("documents", dummy_data)
[{'content': 'The rat the cat the dog bit chased escaped.', 'id': 2}, {'content': 'More people have been to Russia than I have.', 'id': 3}]
```

--------------------------------

### Create and Configure an Agent

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_plugins.html

Instantiate an `Agent` with a chosen LLM, custom instructions, and the defined tool. This example uses `VllmModel` and provides instructions for file access.

```python
from wayflowcore import Agent
from wayflowcore.models import VllmModel

llm = VllmModel(
    model_id="LLAMA_MODEL_ID",
    host_port="LLAMA_API_URL",
)

custom_instructions = """
You have the following files available:
- movies.txt
- videogames.txt

Read the content of the right file - based on the user query - with the tool, and answer the user.
"""

assistant = Agent(
    custom_instruction=custom_instructions,
    tools=[ReadFileTool()],
    llm=llm,
)

```

--------------------------------

### Example PDF Content

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_build_assistants_with_tools.html

This is a sample text content extracted from a PDF document, used for demonstration purposes in the guide.

```text
# Oracle Corporation

Oracle Corporation is an American multinational computer technology company headquartered in Austin, Texas. Co-founded in 1977 in Santa Clara, California, by Larry Ellison, who remains executive chairman, Oracle Corporation is the fourth-largest software company in the world by market capitalization as of 2025. Its market value was approximately US$614 billion in June 2025. The company's 2023 ranking in the Forbes Global 2000 was 80. The company sells database software, (particularly the Oracle Database), and cloud computing software and hardware. Oracle's core application software is a suite of enterprise software products, including enterprise resource planning (ERP), human capital management (HCM), customer relationship management (CRM), enterprise performance management (EPM), Customer Experience Commerce (CX Commerce) and supply chain management (SCM) software.

```

--------------------------------

### Agent Initialization

Source: https://oracle.github.io/wayflow/26.1.1/core/api/agent.html

Demonstrates how to initialize an Agent instance.

```APIDOC
## Agent Class

### Description
Represents an agent that can interact with users and perform tasks.

### Parameters
- **llm** (LlmModel) - The language model to be used by the agent.
- **input_descriptors** (Optional[List[Property]]) - Overrides resolved descriptors, matched by name.
- **output_descriptors** (Optional[List[Property]]) - Outputs the agent is expected to generate.
- **name** (Optional[str]) - Name of the agent for composition.
- **description** (str) - Description of the agent for composition.
- **id** (Optional[str]) - ID of the agent.
- **agent_template** (Optional[PromptTemplate]) - Specific agent template for advanced prompting.
- **transforms** (Optional[List[MessageTransform]]) - List of MessageTransform objects to run on each conversation.

### Example
```python
>>> from wayflowcore.agent import Agent
>>> agent = Agent(llm=llm)
```
```

--------------------------------

### Execute a Flow

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_mcp.html

Run the constructed Flow and retrieve its outputs. This example demonstrates starting a conversation, executing it, and processing the results.

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

--------------------------------

### Execute Flow and Get Summary

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_promptexecutionstep.html

Start the WayFlow conversation, execute the flow, and print the generated summary. Handles different execution statuses.

```python
from wayflowcore.executors.executionstatus import FinishedStatus

conversation = flow.start_conversation(inputs={"article": article})
status = conversation.execute()
if isinstance(status, FinishedStatus):
    print(status.output_values["summary"])
else:
    print(f"Invalid execution status, expected FinishedStatus, received {type(status)}")
# Sea turtles face threats from poaching, habitat loss, and pollution globally.

```

--------------------------------

### Create and run an agent with a server tool

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_build_assistants_with_tools.html

Initializes an Agent assistant using `build_agent`, starts a conversation, appends a user message with a file path, and executes the conversation.

```python
assistant = build_agent(llm, read_pdf_server_tool)

conversation = assistant.start_conversation()

conversation.append_user_message(
    f"Please summarize my PDF document (can be found at {PDF_FILE_PATH})"
)
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)}")
```

--------------------------------

### Create PromptExecutionStep and inspect input descriptors

Source: https://oracle.github.io/wayflow/26.1.1/core/api/flows.html

Demonstrates how to initialize a PromptExecutionStep with an LLM and a prompt template. It also shows how to inspect the automatically inferred input descriptors from the template.

```python
>>> from wayflowcore.steps import PromptExecutionStep
>>> from wayflowcore.flowhelpers import create_single_step_flow
>>> from wayflowcore.models.llmmodelfactory import LlmModelFactory
>>> VLLM_CONFIG = {"model_type": "vllm", "host_port": LLAMA70B_API_ENDPOINT, "model_id": "/storage/models/Llama-3.3-70B-Instruct",}
>>> llm = LlmModelFactory.from_config(VLLM_CONFIG)
>>> step = PromptExecutionStep(
...     prompt_template="What is the capital of {{ country }} The answer is among {% for city in cities %}- {{city}}\n{% endfor %}",
...     llm=llm,
... )
>>> [s.name for s in step.input_descriptors]
['country', 'cities']
```

--------------------------------

### Define Flow with User Input Step

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_userinputinflows.html

Construct a WayFlow flow that includes steps for starting, getting user information via a tool, asking the user a question, and processing the response with an LLM.

```python
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.dataconnection import DataFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.steps import (
   BranchingStep,
   CompleteStep,
   ExtractValueFromJsonStep,
   InputMessageStep,
   PromptExecutionStep,
   StartStep,
   ToolExecutionStep,
)
from wayflowcore.tools import tool

@tool(description_mode="only_docstring")
def get_user_name_tool() -> str:
    """Tool to get user name."""
    return "Alice"

start_step = StartStep(name="start")

get_user_name_step = ToolExecutionStep(
   name="get_user_name_step",
   tool=get_user_name_tool,
)

ask_user_request_step = InputMessageStep(
   name="ask_user_request_step", message_template="Hi {{username}}. What can I do for you today?"
)

answer_request_step = PromptExecutionStep(
   name="answer_request_step",
   llm=llm,
   prompt_template="Your are an helpful assistant. Help answer the user request: {{request}}",
   output_mapping={
       PromptExecutionStep.OUTPUT: "my_output"
   },  # what we want to expose as the output name
)

end_step = CompleteStep(name="end")

flow = Flow(
   begin_step=start_step,
   control_flow_edges=[
       ControlFlowEdge(start_step, get_user_name_step),
       ControlFlowEdge(get_user_name_step, ask_user_request_step),
       ControlFlowEdge(ask_user_request_step, answer_request_step),
       ControlFlowEdge(answer_request_step, end_step),
   ],
   data_flow_edges=[
       DataFlowEdge(
           get_user_name_step, ToolExecutionStep.TOOL_OUTPUT, ask_user_request_step, "username"
       ),
       DataFlowEdge(
           ask_user_request_step,
           InputMessageStep.USER_PROVIDED_INPUT,
           answer_request_step,
           "request",
       ),
   ],
)
```

--------------------------------

### Integrate Transform into Agent and Start Conversation

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_long_context.html

Add the initialized ConversationSummarizationTransform to the Agent's transforms list. This setup allows the agent to automatically summarize the conversation when it exceeds the configured message limit.

```python
agent = Agent(
    llm=agent_llm,
    tools=[],
    transforms=[transform],
)

conversation = agent.start_conversation()
conversation.append_user_message("What is the capital of Switzerland?")
conversation.execute()
conversation.append_user_message("Tell me a fun fact about it.")
conversation.execute()
conversation.append_user_message("What is the population?")
conversation.execute()
conversation.append_user_message("What is the area?")
conversation.execute()
conversation.append_user_message("What languages are spoken there?")
conversation.execute()
# The conversation will be summarized when it exceeds 5 messages
```

--------------------------------

### Create and Execute a Simple Flow

Source: https://oracle.github.io/wayflow/26.1.1/core/api/flows.html

Demonstrates creating a basic flow with start, output, and complete steps, then executing it with user input.

```python
>>> from wayflowcore.flow import Flow
>>> from wayflowcore.steps import OutputMessageStep, InputMessageStep, StartStep, CompleteStep
>>> from wayflowcore.controlconnection import ControlFlowEdge
>>> from wayflowcore.property import StringProperty
>>> USER_NAME = "$message"
>>> start_step = StartStep(name="start_step", input_descriptors=[StringProperty(name=USER_NAME)])
>>> output_step = OutputMessageStep(
...     name="output_step",
...     message_template="Welcome, {{username}}",
...     input_mapping={"username": USER_NAME}
... )
>>> complete_step = CompleteStep(name='complete_step')
>>> flow = Flow(
...     begin_step=start_step,
...     steps=[start_step, output_step, complete_step],
...     control_flow_edges=[
...         ControlFlowEdge(source_step=start_step, destination_step=output_step),
...         ControlFlowEdge(source_step=output_step, destination_step=complete_step),
...     ],
... )
>>> conversation = flow.start_conversation(inputs={USER_NAME: "User"})
>>> status = conversation.execute()
>>> status.output_values
{'output_message': 'Welcome, User'}
```

--------------------------------

### AgentExecutionStep Initialization with DuckDuckGoSearch

Source: https://oracle.github.io/wayflow/26.1.1/core/api/flows.html

Initializes an AgentExecutionStep using a DuckDuckGoSearch agent. This example demonstrates setting up the step with a specific agent and imports necessary components for its creation.

```python
>>> from wayflowcore.flow import Flow
>>> from wayflowcore.agent import Agent
>>> from wayflowcore.property import (
...     IntegerProperty,
...     Property,
...     StringProperty,
... )
>>> from wayflowcore.flow import Flow
>>> from wayflowcore.steps import InputMessageStep
>>> from wayflowcore.steps.agentexecutionstep import AgentExecutionStep, CallerInputMode
>>> from langchain_community.tools import DuckDuckGoSearchRun  
```

--------------------------------

### Build Flow with User Input

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_userinputinflows.html

Construct a WayFlow with an InputMessageStep to prompt the user and subsequent steps to process the input. This example includes a tool to get a username and an LLM step to respond to the user's request.

```python
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.dataconnection import DataFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.steps import (
    BranchingStep,
    CompleteStep,
    ExtractValueFromJsonStep,
    InputMessageStep,
    PromptExecutionStep,
    StartStep,
    ToolExecutionStep,
)
from wayflowcore.tools import tool

@tool(description_mode="only_docstring")
def get_user_name_tool() -> str:
    """Tool to get user name."""
    return "Alice"

start_step = StartStep(name="start")

get_user_name_step = ToolExecutionStep(
    name="get_user_name_step",
    tool=get_user_name_tool,
)

ask_user_request_step = InputMessageStep(
    name="ask_user_request_step", message_template="Hi {{username}}. What can I do for you today?"
)

answer_request_step = PromptExecutionStep(
    name="answer_request_step",
    llm=llm,
    prompt_template="Your are an helpful assistant. Help answer the user request: {{request}}",
    output_mapping={
        PromptExecutionStep.OUTPUT: "my_output"
    },  # what we want to expose as the output name
)

end_step = CompleteStep(name="end")

flow = Flow(
    begin_step=start_step,
    control_flow_edges=[
        ControlFlowEdge(start_step, get_user_name_step),
        ControlFlowEdge(get_user_name_step, ask_user_request_step),
        ControlFlowEdge(ask_user_request_step, answer_request_step),
        ControlFlowEdge(answer_request_step, end_step),
    ],
    data_flow_edges=[
        DataFlowEdge(
            get_user_name_step, ToolExecutionStep.TOOL_OUTPUT, ask_user_request_step, "username"
        ),
        DataFlowEdge(
            ask_user_request_step,
            InputMessageStep.USER_PROVIDED_INPUT,
            answer_request_step,
            "request",
        ),
    ],
)

```

--------------------------------

### Run Swarm-integrated Flow and Get Output

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_swarm.html

This Python code demonstrates how to execute a Wayflow Flow that includes a Swarm agent. It starts a conversation, appends a user message, executes the flow, and prints the final output message.

```python
flow = swarm_in_flow(assistant)
conversation = flow.start_conversation()
conversation.append_user_message("My skin has been itching for about a week. Can you tell me how severe it is?")
status = conversation.execute()
print(status.output_values["output_message"])

```

--------------------------------

### Create and Start a Simple MCP Server

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_mcp.html

This Python script sets up a basic MCP server using FastMCP, exposing 'get_user_session' and 'get_payslips' tools. Run this in a separate Python process.

```python
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

```

--------------------------------

### Create Agent with Tools

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_userconfirmation.html

Initialize an Agent with the configured LLM and a list of tools, including those that require confirmation.

```python
from wayflowcore.agent import Agent

assistant = Agent(
   llm=llm,
   tools=[add_numbers_tool, subtract_numbers_tool, multiply_numbers_tool],
   custom_instruction="Use the tools at your disposal to answer the user requests.",
)
```

--------------------------------

### Define PR Diff Flow Steps

Source: https://oracle.github.io/wayflow/26.1.1/core/tutorials/usecase_prbot.html

Define the WayFlow steps for retrieving and processing a pull request diff. This includes a start step, a tool execution step for getting the diff, and a regex extraction step to parse file diffs.

```python
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.dataconnection import DataFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.property import StringProperty
from wayflowcore.steps import RegexExtractionStep, StartStep, ToolExecutionStep

# IO Variable Names
REPO_DIRPATH_IO = "$repo_dirpath_io"
PR_DIFF_IO = "$raw_pr_diff"
FILE_DIFF_LIST_IO = "$file_diff_list"

# Define the steps

start_step = StartStep(name="start_step", input_descriptors=[StringProperty(name=REPO_DIRPATH_IO)])

# Step 1: Retrieve the pull request diff using the local tool
get_pr_diff_step = ToolExecutionStep(
    name="get_pr_diff",
    tool=local_get_pr_diff_tool,
    raise_exceptions=True,
    input_mapping={"repo_dirpath": REPO_DIRPATH_IO},
    output_mapping={ToolExecutionStep.TOOL_OUTPUT: PR_DIFF_IO},
)

# Step 2: Extract the file diffs from the raw diff using a regular expression
extract_into_list_of_file_diff_step = RegexExtractionStep(
    name="extract_into_list_of_file_diff",
    regex_pattern=r"(diff --git[\s\S]*?)(?=diff --git|$)",
    return_first_match_only=False,
    input_mapping={RegexExtractionStep.TEXT: PR_DIFF_IO},
    output_mapping={RegexExtractionStep.OUTPUT: FILE_DIFF_LIST_IO},
)

# Define the sub flow
```

--------------------------------

### Wayflow Sub-flow Definition

Source: https://oracle.github.io/wayflow/26.1.1/_downloads/851f997ab457f57a59b6c42bfde5d608/usecase_prbot.ipynb

This Python code defines the initial steps and data flow for a Wayflow sub-flow. It sets up a starting step, a tool execution step to get the Git diff, and prepares for a regex extraction step, defining input and output variable names.

```python
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.dataconnection import DataFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.property import StringProperty
from wayflowcore.steps import RegexExtractionStep, StartStep, ToolExecutionStep

# Step Names
START_STEP = "start_step"
GET_PR_DIFF_STEP = "get_pr_diff"
EXTRACT_INTO_LIST_OF_FILE_DIFF_STEP = "extract_into_list_of_file_diff"

# IO Variable Names
REPO_DIRPATH_IO = "$repo_dirpath_io"
PR_DIFF_IO = "$raw_pr_diff"
FILE_DIFF_LIST_IO = "$file_diff_list"

# Define the steps

start_step = StartStep(input_descriptors=[StringProperty(name=REPO_DIRPATH_IO)])

# Step 1: Retrieve the pull request diff using the local tool
get_pr_diff_step = ToolExecutionStep(
    tool=local_get_pr_diff_tool,
    raise_exceptions=True,
    input_mapping={"repo_dirpath": REPO_DIRPATH_IO},
    output_mapping={ToolExecutionStep.TOOL_OUTPUT: PR_DIFF_IO},
)

# Step 2: Extract the file diffs from the raw diff using a regular expression

```

--------------------------------

### Create Dummy Data and Populate Datastore

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_datastores.html

This snippet defines a list of dummy product data and then uses the datastore's 'create' method to populate the 'products' collection. This method supports both bulk and single-element creation.

```python
dummy_data = [
    {
        "ID": 0,
        "title": "Broccoli",
        "description": "Healthy and delicious cruciferous vegetable!",
        "price": 1.5,
        "category": "Produce",
    },
    {
        "ID": 1,
        "title": "Oranges",
        # We introduce a typo in this entity for the Assistant to fix
        "description": "Vitamin C-filled cirus fruits",
        "price": 1.8,
        "category": "Produce",
    },
    {
        "ID": 2,
        "title": "Shampoo",
        "description": "Shiny smooth hair in just 10 applications!",
        "price": 4.5,
        "category": "Personal hygiene",
    },
    {
        "ID": 3,
        "title": "Crushed ice",
        "description": "Cool any drink in seconds!",
        "price": 4.5,
        "category": "Food",
    },
]

# Create supports both bulk-creation, as well as single element creation
datastore.create(collection_name="products", entities=dummy_data)
```

--------------------------------

### Install WayFlowCore

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/howto_a2aagent.html

Set up a Python virtual environment and install the WayFlowCore library. Ensure pip is upgraded before installation.

```bash
python -m venv venv-wayflowcore
source venv-wayflowcore/bin/activate
pip install --upgrade pip
pip install "wayflowcore==26.1.1"
```

--------------------------------

### PromptExecutionStep Initialization

Source: https://oracle.github.io/wayflow/26.1.1/core/api/flows.html

Illustrates the initialization of a PromptExecutionStep, which is used for generating text with an LLM using a prompt template. It highlights default behaviors for input and output descriptors and the option for structured generation.

```python
class PromptExecutionStep(PromptExecutionStep):
    pass
    # ... other parameters like prompt_template, llm, generation_config, etc.
```

--------------------------------

### Create the Agent

Source: https://oracle.github.io/wayflow/26.1.1/core/tutorials/basic_agent.html

Initializes an Agent instance with custom instructions, available tools, and a language model.

```python
assistant = Agent(
    custom_instruction=HRASSISTANT_GENERATION_INSTRUCTIONS,
    tools=[search_hr_database],  # this is a decorated python function (Server tool in this example)
    llm=llm,  # the LLM object we created above
)
```

--------------------------------

### Install OCI Package

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/embeddingmodels_from_different_providers.html

Ensure the `oci` package is installed with version `2.134.0` or higher. Activate your WayFlow environment before installation.

```bash
pip install oci>=2.134.0
```

--------------------------------

### Multimodal Prompt Example

Source: https://oracle.github.io/wayflow/26.1.1/core/api/conversation.html

Demonstrates creating a multimodal prompt with text and image content for LLM generation. Ensure the image content is properly formatted.

```python
>>> prompt = Prompt(messages=[Message(contents = [TextContent("Which company's logo is this?") , img_content])])
>>> completion = multimodal_llm.generate(prompt)
>>> # LlmCompletion(message=Message(content="That is the logo for **Oracle Corporation**."))
```

--------------------------------

### Run Example Script

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/generation_config.html

Execute the example Python script. This demonstrates the usage of WayFlow components.

```bash
python example_generationconfig.py
```

--------------------------------

### Initialize InMemoryDatastore with Entities

Source: https://oracle.github.io/wayflow/26.1.1/core/api/flows.html

Shows how to set up an in-memory datastore with a defined entity schema, including properties like 'id' and 'content'.

```python
>>> from wayflowcore.datastore import Entity
>>> from wayflowcore.datastore.inmemory import InMemoryDatastore
>>> from wayflowcore.flowhelpers import create_single_step_flow
>>> from wayflowcore.property import StringProperty, IntegerProperty
>>> from wayflowcore.steps.datastoresteps import DatastoreCreateStep

>>> document = Entity(
...     properties={ "id": IntegerProperty(), "content": StringProperty(default_value="Empty...") }
... )
>>> datastore = InMemoryDatastore({"documents": document})
```

--------------------------------

### Install WayFlow Core

Source: https://oracle.github.io/wayflow/26.1.1/core/index.html

Installs the wayflowcore package for Python 3.10 from PyPI. Ensure you are using a compatible Python version.

```bash
pip install "wayflowcore==26.1.1"
```

--------------------------------

### Define and Integrate a Map-Reduce Subflow with FlowExecutionStep

Source: https://oracle.github.io/wayflow/26.1.1/core/api/flows.html

This example demonstrates defining reusable tools, creating a map-reduce subflow using PromptExecutionStep and ToolExecutionStep, and then integrating this subflow into a main flow using FlowExecutionStep. It requires defining input/output variables and a prompt template.

```python
from typing import List
from wayflowcore.property import Property, StringProperty, ListProperty, IntegerProperty
from wayflowcore.flowhelpers import create_single_step_flow
from wayflowcore.controlconnection import ControlFlowEdge
from wayflowcore.flow import Flow
from wayflowcore.steps import MapStep, PromptExecutionStep, ToolExecutionStep
from wayflowcore.tools import tool

@tool(description_mode="only_docstring")
def duplication_tool(element: str, n: int) -> List[str]:
    '''Returns a list containing the passed element, n times'''
    return [element for _ in range(n)]

@tool(description_mode="only_docstring")
def reduce_tool(elements: List[str]) -> str:
    '''Returns the average of the first number found in each element'''
    import re
    extracted_elements = [re.search('-?\d*\.?\d+', elt) for elt in elements]
    extracted_numbers = [float(x.group(0)) for x in extracted_elements if x is not None] or [-1.]
    return str(sum(extracted_numbers) / len(extracted_numbers))

# Defining flow input/output variables
USER_QUERY_IO = "$user_query"
N_REPEAT_IO = "$n_repeat"
FLOW_ITERABLE_QUERIES_IO = "$flow_iterable_queries"
FLOW_PROCESSED_QUERIES_IO = "$flow_processed_queries"
FINAL_ANSWER_IO = "$answer_io"
# Defining a simple prompt
REASONING_PROMPT_TEMPLATE = '''Provide your best numerical estimate for: {{user_input}}
Your answer should be a single number. Do not include any units, reasoning, or extra text.'''
# Defining the subflow
duplication_step = ToolExecutionStep(
    tool=duplication_tool,
    input_mapping={"element": USER_QUERY_IO, "n": N_REPEAT_IO},
    output_mapping={ToolExecutionStep.TOOL_OUTPUT: FLOW_ITERABLE_QUERIES_IO},
    name="DUPLICATION",
)
map_step = MapStep(
    flow=create_single_step_flow(
        PromptExecutionStep(
            prompt_template=REASONING_PROMPT_TEMPLATE,
            llm=llm,
            output_mapping={PromptExecutionStep.OUTPUT: FLOW_PROCESSED_QUERIES_IO},
        ),
        step_name="REASONING"
    ),
    unpack_input={"user_input": "."},
    output_descriptors=[ListProperty(name=FLOW_PROCESSED_QUERIES_IO, item_type=StringProperty())],
    input_mapping={MapStep.ITERATED_INPUT: FLOW_ITERABLE_QUERIES_IO},
    name="MAP"
)
reduce_step = ToolExecutionStep(
    tool=reduce_tool,
    input_mapping={"elements": FLOW_PROCESSED_QUERIES_IO},
    output_mapping={ToolExecutionStep.TOOL_OUTPUT: FINAL_ANSWER_IO},
    name="REDUCE",
)
mapreduce_flow = Flow.from_steps(steps=[duplication_step, map_step, reduce_step])

```

--------------------------------

### Initialize HumanProxyAssistant

Source: https://oracle.github.io/wayflow/26.1.1/core/api/evaluation.html

Instantiate HumanProxyAssistant with LLM, task descriptions, and role information. The `extra_instructions` parameter can be used to guide the LLM's output format.

```python
>>> from wayflowcore.evaluation import HumanProxyAssistant
>>> human_proxy = HumanProxyAssistant(
...     llm=llm,
...     full_task_description="to provide you with the current general weather (breezy, warm, snowy, etc.) in the largest city of Switzerland (don't tell me which one).",
...     short_task_description="get the weather in the largest city of Switzerland",
...     assistant_role="a weather reporter",
...     user_role="a news viewer"
... )
```

--------------------------------

### Start and Execute Conversation

Source: https://oracle.github.io/wayflow/26.1.1/core/howtoguides/agents.html

Starts a conversation with the agent, executes it, and retrieves the last message. Handles cases where no message is returned.

```python
conversation = agent.start_conversation()
conversation.execute()
last_message = conversation.get_last_message()
print(last_message.content if last_message else "No message")
# Welcome to our article writing guide. How can I assist you today?
```

--------------------------------

### Define and Execute a Simple Flow

Source: https://oracle.github.io/wayflow/26.1.1/core/api/flows.html

Demonstrates creating a flow with a mapreduce step and an output step, then starting a conversation and executing it. Ensure necessary imports and constants are defined.

```python
from wayflowcore.flow import Flow
from wayflowcore.steps import FlowExecutionStep, OutputMessageStep
MAPREDUCE_STEP = "MAPREDUCE"
OUTPUT_STEP = "OUTPUT"
mapreduce_step = FlowExecutionStep(mapreduce_flow, name=MAPREDUCE_STEP)
output_step = OutputMessageStep(
    "The estimation is {{value}}",
    input_mapping={"value": FINAL_ANSWER_IO},
    name=OUTPUT_STEP,
)
assistant = Flow(
    begin_step=mapreduce_step,
    control_flow_edges=[
        ControlFlowEdge(source_step=mapreduce_step, destination_step=output_step),
        ControlFlowEdge(source_step=output_step, destination_step=None),
    ],
)
conversation = assistant.start_conversation(inputs={
    USER_QUERY_IO: "How many calories are in a typical slice of pepperoni pizza?",
    N_REPEAT_IO: 2
})
status = conversation.execute()
# print(conversation.get_last_message().content)
# The estimation is 285.5
```

--------------------------------

### Agent Spec JSON Structure Example

Source: https://oracle.github.io/wayflow/26.1.1/core/tutorials/basic_flow.html

This is an example of the JSON structure generated by AgentSpecExporter. It defines the components, connections, and metadata of an agent flow.

```json
{
  "component_type": "Flow",
  "id": "bf3ca5b3-39bc-40ca-9d9a-a387bee07031",
  "name": "flow_e057bcd6__auto",
  "description": "",
  "metadata": {
    "__metadata_info__": {}
  },
  "inputs": [],
  "outputs": [
    {
      "description": "the input value provided by the user",
      "type": "string",
      "title": "user_query"
    },
    {
      "type": "string",
      "title": "hr_data_context"
    },
    {
      "type": "string",
      "title": "answer"
    },
    {
      "description": "the message added to the messages list",
      "type": "string",
      "title": "output_message"
    }
  ],
  "start_node": {
    "$component_ref": "91594dfc-44fa-41d9-91ba-52ae7b0b0f16"
  },
  "nodes": [
    {
      "$component_ref": "91594dfc-44fa-41d9-91ba-52ae7b0b0f16"
    },
    {
      "$component_ref": "c79bea59-c54b-4a54-8ac5-df6daaa60d22"
    },
    {
      "$component_ref": "819ae9c0-4055-4fe6-ab05-f03c45775206"
    },
    {
      "$component_ref": "d9b795cd-3949-4acb-9399-6d9f0a10d7e6"
    },
    {
      "$component_ref": "01c4a5ff-6568-42a0-a8de-8efaec780712"
    },
    {
      "$component_ref": "953ccf06-5550-4597-9518-df1043269693"
    }
  ],
  "control_flow_connections": [
    {
      "component_type": "ControlFlowEdge",
      "id": "a71aebf8-0416-4811-bb89-6cf4627dc170",
      "name": "start_step_to_user_input_step_control_flow_edge",
      "description": null,
      "metadata": {
        "__metadata_info__": {}
      },
      "from_node": {
        "$component_ref": "91594dfc-44fa-41d9-91ba-52ae7b0b0f16"
      },
      "from_branch": null,
      "to_node": {
        "$component_ref": "c79bea59-c54b-4a54-8ac5-df6daaa60d22"
      }
    },
    {
      "component_type": "ControlFlowEdge",
      "id": "b2e0fb5c-8b57-4c32-9261-bd036ee607c1",
      "name": "user_input_step_to_hr_lookup_step_control_flow_edge",
      "description": null,
      "metadata": {
        "__metadata_info__": {}
      },
      "from_node": {
        "$component_ref": "c79bea59-c54b-4a54-8ac5-df6daaa60d22"
      },
      "from_branch": null,
      "to_node": {
        "$component_ref": "819ae9c0-4055-4fe6-ab05-f03c45775206"
      }
    },
    {
      "component_type": "ControlFlowEdge",
      "id": "4d1bea5b-0c63-4042-891b-aca964e1640f",
      "name": "hr_lookup_step_to_llm_answer_step_control_flow_edge",
      "description": null,
      "metadata": {
        "__metadata_info__": {}
      },
      "from_node": {
        "$component_ref": "819ae9c0-4055-4fe6-ab05-f03c45775206"
      },
      "from_branch": null,
      "to_node": {
        "$component_ref": "d9b795cd-3949-4acb-9399-6d9f0a10d7e6"
      }
    },
    {
      "component_type": "ControlFlowEdge",
      "id": "9323cd7b-5da2-46b6-8333-46ac19d22d4d",
      "name": "llm_answer_step_to_user_output_step_control_flow_edge",
      "description": null,
      "metadata": {
        "__metadata_info__": {}
      },
      "from_node": {
        "$component_ref": "d9b795cd-3949-4acb-9399-6d9f0a10d7e6"
      },
      "from_branch": null,
      "to_node": {
        "$component_ref": "01c4a5ff-6568-42a0-a8de-8efaec780712"
      }
    },
    {
      "component_type": "ControlFlowEdge",
      "id": "e8075a13-3e48-4611-b68f-aa10423309ec",
      "name": "user_output_step_to_final_step_control_flow_edge",
      "description": null,
      "metadata": {
        "__metadata_info__": {}
      },
      "from_node": {
        "$component_ref": "01c4a5ff-6568-42a0-a8de-8efaec780712"
      },
      "from_branch": null,
      "to_node": {
        "$component_ref": "953ccf06-5550-4597-9518-df1043269693"
      }
    }
  ],
  "data_flow_connections": [
    {
      "component_type": "DataFlowEdge",
      "id": "1a76f7f9-19b0-438e-aa3e-274dc06d4aa2",
      "name": "user_input_step_user_query_to_hr_lookup_step_query_data_flow_edge",
      "description": null,
      "metadata": {
        "__metadata_info__": {}
      },
      "source_node": {
        "$component_ref": "c79bea59-c54b-4a54-8ac5-df6daaa60d22"
      },
      "source_output": "user_query",
      "destination_node": {
        "$component_ref": "819ae9c0-4055-4fe6-ab05-f03c45775206"
      },
      "destination_input": "query"
    },
    {
      "component_type": "DataFlowEdge",
      "id": "f4a11a6e-0b79-4ede-8673-30ad1b79af2d",
      "name": "user_input_step_user_query_to_llm_answer_step_user_question_data_flow_edge",
      "description": null,
      "metadata": {
        "__metadata_info__": {}
      },
      "source_node": {
        "$component_ref": "c79bea59-c54b-4a54-8ac5-df6daaa60d22"
      },
      "source_output": "user_query",
      "destination_node": {

```
Tokens:
Raw