Integrating External Services with Temporal

Community Support
1

Did i get this right?

The Temporal brings the maximum value with this approach. It serves as a service mesh for invocations of child workflows and activities hosted by different services. All invocations are flow controlled and operations can be long running. It also has fewer moving parts and is more secure as no direct RPC communications are happening.
How to let one workflow interact with multiple microservices

External API Calls better to make this way with Temporal?
            +-------------------+
            |   Main Workflow   |
            +-------------------+
                    |
                    v
            +-----------------+
            | Parent Workflow |
            +-----------------+
                    |
       +--------------------------+
       |         Triggers         |
       v                          v
+-----------------+        +-----------------+
| Service A       |        | Service B       |
| Workflow A      |        | Workflow B      |
+-----------------+        +-----------------+
       |                             |
       +-----------------------------+
                    |
      +-----------------------------+
      |     API Gateway Service     |
      +-----------------------------+
                    |
      +-----------------------------+
      |      External API Calls     |
      +-----------------------------+
Code Example:
# Main Workflow
class MainWorkflow:
    @workflow.defn
    def run(self):
        parent_workflow = workflow.execute_child_workflow(ParentWorkflow.run)
        workflow.wait_for(parent_workflow)

# Parent Workflow
class ParentWorkflow:
    @workflow.defn
    def run(self):
        workflow_a = workflow.execute_child_workflow(WorkflowA.run)
        workflow_b = workflow.execute_child_workflow(WorkflowB.run)
        workflow.wait_for_all([workflow_a, workflow_b])

# Service A Workflow
class WorkflowA:
    @workflow.defn
    def run(self):
        activity_result = activities.call(ActivityForServiceA.run)
        # Handle the response

# Activity for Service A to call an external API
class ActivityForServiceA:
    @activity.defn
    def run(self):
        # Call external API, e.g., OpenAI API
        response = call_openai_api()
        return response

# Function to call OpenAI API
def call_openai_api():
    import requests
    response = requests.post("https://api.openai.com/v1/completions", json={...})
    return response.json()

Thanks!

2

If WorkflowA calls a single activity, you don’t need it. Just call the activity directly from the parent workflow.

1 Like