Non-blocking agent functions with Temporal?

[darugar]

Is there an example of a function (as in a function available to an agent) that does not block temporal workflows?

I have a function that runs a commandline utility (whois) to check if a domain is registered. I'm making it available to the agent workflow via functions. This causes the workflow to run into the following issue: [TMPRL1101] Potential deadlock detected: workflow didn't yield within 2 second(s).

How do I create a a non-blocking / deadlock causing version of this function?

For now I've converted it to an MCP server, but it seems painful to make every utility MCP.

def is_domain_registered(domain: str) -> bool:
...
        result = subprocess.run(
            ["whois", domain],
            capture_output=True,
            text=True,
            timeout=10
        )
...
        return is_registered

...

    @app.workflow_run
    async def run(
...
        my_agent = Agent(
            name=data.slug,
            instruction=data.instruction,
            server_names=data.server_names,
            functions=[is_domain_registered,]
        )

[saqadri]

@darugar thanks for reporting!

What you are describing should be possible, though I need to double check if the functions will do the right thing within the agent.
One thing you should consider is marking is_domain_registered with @app.workflow_task. Functions marked with that will get executed as activities.

To be super sure, you could create a helper function

def is_domain_registered_impl(domain: str) -> bool:
...
        result = subprocess.run(
            ["whois", domain],
            capture_output=True,
            text=True,
            timeout=10
        )
...
        return is_registered

def is_domain_registered(domain: str, context: Context) -> bool:
    executor = context.executor
    result = await executor.execute_activity(is_domain_registered, domain)
    return result

This will guarantee that we run the function as an activity.

BTW, we have a few new syntactic sugar approaches that may make it a lot easier to define workflows. Check out the @app.tool and @app.async_tool decorators in this example:

https://github.com/lastmile-ai/mcp-agent/blob/main/examples/mcp_agent_server/temporal/main.py#L96-100

This automatically sets up the function to be run as a temporal workflow. You can again use the executor.execute_activity function to run sub-functions as activities, or decorate using @app.workflow_task. Just FYI if that makes your code simpler.

Please let me know if the above helps, and if not happy to pair with you and investigate more carefully.

[darugar]

This doesn't seem to work for me. Details on what I tried below.

Would it be possible to include a full example of an async function with an agent? I think these are useful because agents will want to access tools that are not instantaneous. I'm happy to contribute the example if we get this working.

---

The simple non-async case works:

def twaste(a: int, b: int) -> int:
    """
    twaste two numbers
    """
    print(f"twasting {a} and {b}")
    return (a * b)-3


finder_agent = Agent(
    name=data.slug,
    instruction=data.instruction,
    server_names=data.server_names,
    functions=[twaste]

Simple non-async, with delay:

def twaste(a: int, b: int) -> int:
    """
    twaste two numbers
    """
    print(f"twasting {a} and {b}")
    time.sleep(6)
    return (a * b)-3

I get the following error (as expected):

[TMPRL1101] Potential deadlock detected: workflow didn't yield within 2 second(s).

With @app.workflow_task:

@app.workflow_task
def twaste(a: int, b: int) -> int:
    """
    twaste two numbers
    """
    print(f"twasting {a} and {b}")
    time.sleep(6)
    return (a * b)-3

Error:

Cannot generate a JsonSchema for core_schema.CallableSchema

For further information visit https://errors.pydantic.dev/2.11/u/invalid-for-json-schema

With executor.execute_activity:

def twaste_impl(a: int, b: int) -> int:
    """
    twaste two numbers
    """
    print(f"twasting {a} and {b}")
    time.sleep(6)
    return (a * b)-3

async def twaste(a: int, b: int, context: Context) -> bool:
    """
    twaste two numbers
    """
    executor = context.executor
    result = await executor.execute_activity(twaste_impl, a, b)
    return result

Error:

Cannot generate a JsonSchema for core_schema.IsInstanceSchema (<class 'mcp_agent.executor.executor.Executor'>)

For further information visit https://errors.pydantic.dev/2.11/u/invalid-for-json-schema

With executor.execute_activity, without async (your example had an await but no async on the function definition):

def twaste(a: int, b: int, context: Context) -> bool:
    """
    twaste two numbers
    """
    executor = context.executor
    result = executor.execute_activity(twaste_impl, a, b)
    return result

Error:

Cannot generate a JsonSchema for core_schema.IsInstanceSchema (<class 'mcp_agent.executor.executor.Executor'>)

For further information visit https://errors.pydantic.dev/2.11/u/invalid-for-json-schema

I also tried with temporal @activity.defn:

@activity.defn
def twaste(a: int, b: int) -> int:
    """
    twaste two numbers
    """
    print(f"twasting {a} and {b}")
    time.sleep(6)
    return (a * b)-3

[TMPRL1101] Potential deadlock detected: workflow didn't yield within 2 second(s).