> ## Documentation Index
> Fetch the complete documentation index at: https://docs.mcp-agent.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Workflows and Decorators

> Understanding the Workflow class and decorator-based tool definition in mcp-agent

## Overview

mcp-agent gives you two complementary ways to expose agent behaviour:

1. **Decorator-based tools** – mark a plain Python function with `@app.tool` or `@app.async_tool` to expose it as an MCP tool. This is the quickest way to add synchronous or long-running behaviour to your app.
2. **Workflow classes** – build stateful, structured flows by subclassing `Workflow[T]`. Workflows give you fine-grained control over orchestration, retries, and Temporal integration.

Both options register MCP tools automatically, so any MCP client can invoke them. The high-level “workflow patterns” in [`examples/workflows`](https://github.com/lastmile-ai/mcp-agent/tree/main/examples/workflows) (parallel, router, orchestrator, etc.) are built using these same primitives—they are patterns, not the `Workflow` base class itself.

The rest of this page walks through the decorators first (because most apps start there) and then dives into the `Workflow` class.

## Decorator-based tools

### `@app.tool` – synchronous tools

Use `@app.tool` when the work can complete within a single MCP call. The return value is sent straight back to the client—no polling required.

```python theme={null}
from mcp_agent.app import MCPApp
from typing import Optional

app = MCPApp(name="utility_agent")

@app.tool
async def calculate_sum(numbers: List[float]) -> float:
    """Calculate the sum of a list of numbers."""
    return sum(numbers)

@app.tool(name="get-weather")
async def get_weather(
    city: str,
    units: str = "celsius",
    app_ctx: Optional[Context] = None,
) -> dict:
    if app_ctx:
        app_ctx.logger.info("Fetching weather", data={"city": city})
    return await fetch_weather_api(city, units)
```

Key points:

* Works great for quick operations or simple glue code.
* You can accept an optional `app_ctx: Context` parameter to access logging, server registry, etc.
* The tool result is serialised and returned to the caller immediately.

### `@app.async_tool` – long-running tools

Agents often need to run tasks that take longer than an MCP request allows (multi-step research, human-in-the-loop flows, durable Temporal runs). Decorate those entry points with `@app.async_tool`:

```python theme={null}
@app.async_tool(name="analyze-document")
async def analyze_document_async(
    document_url: str,
    analysis_type: str = "summary",
    app_ctx: Optional[Context] = None,
) -> dict:
    workflow = DocumentAnalysisWorkflow()
    handle = await app_ctx.executor.start_workflow(
        workflow,
        {"url": document_url, "type": analysis_type},
    )
    return {"workflow_id": workflow.id, "run_id": handle.id}
```

`@app.async_tool` starts a workflow in the background and returns identifiers that clients can poll via the built-in `workflows-get_status` tool. This pattern keeps your agent responsive even when the underlying work takes minutes or requires human decisions.

> **Tip:** Agent servers rely heavily on these decorators—see [Agent Servers](/mcp-agent-sdk/mcp/agent-as-mcp-server) for end-to-end examples.

## The Workflow Class

The `Workflow[T]` base class lets you model multi-step or stateful logic while still exposing an MCP tool. Workflows are most useful when you need retries, shared state, or tight integration with the execution engine (asyncio or Temporal).

### Basic workflow definition

```python theme={null}
from mcp_agent.executor.workflow import Workflow, WorkflowResult

# Assume `read_file` / `summarise` are helper functions you provide.

@app.workflow
class SummariseFile(Workflow[str]):
    @app.workflow_run
    async def run(self, path: str) -> WorkflowResult[str]:
        content = await read_file(path)
        summary = await summarise(content)
        return WorkflowResult(value=summary)
```

Decorate the class with `@app.workflow` and the entry point with `@app.workflow_run`. Whatever you return from the method becomes the MCP tool result.

### Useful workflow features

* Access `self.context` for logging, MCP connections, and configuration.
* Store reusable helpers or caches on `self` inside `__init__`.
* Raise exceptions to trigger retries (Temporal) or propagate errors to the caller.
* Combine with `@app.workflow_task` / `@app.workflow_signal` when you need durable activities or signal handlers.

See the sections below for more elaborate compositions.

## Workflow patterns (examples/workflows)

The repository has an [`examples/workflows`](https://github.com/lastmile-ai/mcp-agent/tree/main/examples/workflows) directory that demonstrates higher-level agent patterns: router, parallel fan-out, orchestrator, evaluator/optimizer, and more. These samples compose agents and AugmentedLLMs with helpers from `mcp_agent.workflows.factory`. They do **not** correspond one-to-one with the `Workflow` base class above—they are ready-made orchestration patterns you can adopt or customise.

Use the patterns when you want opinionated orchestration, and drop down to the `Workflow` class (or `@app.async_tool`) when you need bespoke control flow.

## Advanced Workflow Patterns

### Workflow Composition

Compose complex workflows from simpler ones:

```python theme={null}
@app.workflow
class CompositeWorkflow(Workflow[dict]):
    @app.workflow_run
    async def run(self, request: dict) -> WorkflowResult[dict]:
        # Run sub-workflows
        step1 = DataFetchWorkflow()
        data = await step1.run(request["source"])
        
        step2 = DataProcessWorkflow()
        processed = await step2.run(data.value)
        
        step3 = ReportGenerationWorkflow()
        report = await step3.run(processed.value)
        
        return WorkflowResult(value={
            "data": data.value,
            "processed": processed.value,
            "report": report.value
        })
```

### Workflow with Agents

Integrate agents into workflows:

```python theme={null}
from mcp_agent.agents.agent import Agent
from mcp_agent.workflows.llm.augmented_llm_openai import OpenAIAugmentedLLM

@app.workflow
class AgentWorkflow(Workflow[str]):
    @app.workflow_run
    async def run(self, task: str) -> WorkflowResult[str]:
        # Create specialized agent
        agent = Agent(
            name="researcher",
            instruction="Research thoroughly and provide detailed analysis.",
            server_names=["fetch", "filesystem"]
        )
        
        async with agent:
            # Attach LLM
            llm = await agent.attach_llm(OpenAIAugmentedLLM)
            
            # Execute task
            result = await llm.generate_str(task)
            
            return WorkflowResult(value=result)
```

### Parallel Workflow Execution

Execute multiple workflows in parallel:

```python theme={null}
import asyncio

@app.workflow
class ParallelWorkflow(Workflow[dict]):
    @app.workflow_run
    async def run(self, tasks: List[str]) -> WorkflowResult[dict]:
        # Create workflow instances
        workflows = [
            TaskWorkflow() for _ in tasks
        ]
        
        # Run in parallel
        results = await asyncio.gather(*[
            w.run(task) for w, task in zip(workflows, tasks)
        ])
        
        # Combine results
        combined = {
            f"task_{i}": r.value 
            for i, r in enumerate(results)
        }
        
        return WorkflowResult(value=combined)
```

### Stateful Workflows

Maintain state across workflow executions:

```python theme={null}
@app.workflow
class StatefulWorkflow(Workflow[dict]):
    def __init__(self):
        super().__init__()
        self.state = {}
    
    @app.workflow_run
    async def run(self, action: dict) -> WorkflowResult[dict]:
        action_type = action.get("type")
        
        if action_type == "set":
            self.state[action["key"]] = action["value"]
            return WorkflowResult(value={"status": "set"})
        
        elif action_type == "get":
            value = self.state.get(action["key"])
            return WorkflowResult(value={"value": value})
        
        elif action_type == "clear":
            self.state.clear()
            return WorkflowResult(value={"status": "cleared"})
        
        return WorkflowResult(value=self.state)
```

## Temporal Integration

Workflows seamlessly support Temporal for durable execution:

```python theme={null}
# Configure for Temporal
app = MCPApp(
    name="temporal_agent",
    settings=Settings(
        execution_engine="temporal",
        temporal=TemporalSettings(
            host="localhost",
            port=7233,
            namespace="default",
            task_queue="mcp-agent"
        )
    )
)

@app.workflow
class DurableWorkflow(Workflow[str]):
    @app.workflow_run
    async def run(self, task: str) -> WorkflowResult[str]:
        # This workflow is now durable
        # It can be paused, resumed, and retried
        
        # Wait for signal (human-in-the-loop)
        await app.context.executor.signal_bus.wait_for_signal(
            Signal(name="approve", workflow_id=self.id)
        )
        
        # Continue after approval
        result = await self.process_with_approval(task)
        return WorkflowResult(value=result)
```

## MCP Server Integration

### Exposing Workflows as MCP Tools

Workflows and tools are automatically exposed when creating an MCP server:

```python theme={null}
from mcp_agent.mcp.server import create_mcp_server_for_app

# Define workflows and tools
@app.workflow
class MyWorkflow(Workflow[str]):
    @app.workflow_run
    async def run(self, input: str) -> WorkflowResult[str]:
        return WorkflowResult(value=f"Processed: {input}")

@app.tool
async def my_tool(param: str) -> str:
    return f"Tool result: {param}"

# Create MCP server
async def main():
    async with app.run():
        mcp_server = create_mcp_server_for_app(app)
        
        # Available tools:
        # - workflows-list
        # - workflows-MyWorkflow-run
        # - workflows-get_status
        # - my_tool
        
        await mcp_server.run_stdio_async()
```

### Tool Discovery

MCP clients can discover available tools:

```python theme={null}
# From MCP client perspective
tools = await server.list_tools()
for tool in tools:
    print(f"Tool: {tool.name}")
    print(f"Description: {tool.description}")
    print(f"Parameters: {tool.input_schema}")
```

## Best Practices

<AccordionGroup>
  <Accordion title="Choose the Right Abstraction">
    * Use `@app.tool` for simple, stateless operations
    * Use `@app.async_tool` for long-running operations that need polling
    * Use `Workflow` class for complex, multi-step processes
  </Accordion>

  <Accordion title="Type Hints and Documentation">
    Always provide type hints and docstrings:

    ```python theme={null}
    @app.tool
    async def process_data(
        data: dict,
        options: Optional[dict] = None
    ) -> dict:
        """
        Process data with optional transformations.
        
        Args:
            data: Input data to process
            options: Optional processing options
            
        Returns:
            Processed data dictionary
        """
        # Implementation
    ```
  </Accordion>

  <Accordion title="Error Handling">
    Handle errors gracefully:

    ```python theme={null}
    @app.workflow
    class SafeWorkflow(Workflow[str]):
        @app.workflow_run
        async def run(self, input: str) -> WorkflowResult[str]:
            try:
                result = await self.process(input)
                return WorkflowResult(value=result)
            except Exception as e:
                logger.error(f"Processing failed: {e}")
                return WorkflowResult(
                    value=None,
                    error=str(e)
                )
    ```
  </Accordion>

  <Accordion title="Resource Management">
    Use context managers for resources:

    ```python theme={null}
    @app.workflow
    class ResourceWorkflow(Workflow[str]):
        @app.workflow_run
        async def run(self, query: str) -> WorkflowResult[str]:
            async with self.get_database() as db:
                result = await db.query(query)
                return WorkflowResult(value=result)
    ```
  </Accordion>

  <Accordion title="Logging and Observability">
    Use structured logging:

    ```python theme={null}
    @app.tool
    async def monitored_tool(input: str, app_ctx: Optional[Context] = None) -> str:
        if app_ctx:
            logger = app_ctx.logger
            logger.info("Tool started", data={"input": input})
            
            try:
                result = await process(input)
                logger.info("Tool completed", data={"result_length": len(result)})
                return result
            except Exception as e:
                logger.error("Tool failed", data={"error": str(e)})
                raise
    ```
  </Accordion>
</AccordionGroup>

## Testing Workflows

Test your workflows locally:

```python theme={null}
import asyncio
import pytest

@pytest.mark.asyncio
async def test_workflow():
    app = MCPApp(name="test_app")
    
    @app.workflow
    class TestWorkflow(Workflow[str]):
        @app.workflow_run
        async def run(self, input: str) -> WorkflowResult[str]:
            return WorkflowResult(value=input.upper())
    
    async with app.run():
        workflow = TestWorkflow()
        result = await workflow.run("hello")
        assert result.value == "HELLO"
```

## Migration Guide

### From Functions to Tools

```python theme={null}
# Before: Plain function
async def calculate(x: int, y: int) -> int:
    return x + y

# After: MCP tool
@app.tool
async def calculate(x: int, y: int) -> int:
    """Calculate sum of two numbers."""
    return x + y
```

### From Scripts to Workflows

```python theme={null}
# Before: Script
async def main():
    data = await fetch_data()
    processed = await process_data(data)
    await save_results(processed)

# After: Workflow
@app.workflow
class DataPipeline(Workflow[dict]):
    @app.workflow_run
    async def run(self, source: str) -> WorkflowResult[dict]:
        data = await self.fetch_data(source)
        processed = await self.process_data(data)
        await self.save_results(processed)
        return WorkflowResult(value=processed)
```

## Next Steps

<CardGroup cols={2}>
  <Card title="Workflow Patterns" icon="diagram-project" href="/workflows/overview">
    Explore pre-built workflow patterns
  </Card>

  <Card title="Agent Server" icon="server" href="/cloud/agent-server">
    Deploy workflows as MCP servers
  </Card>

  <Card title="Temporal Integration" icon="clock" href="/advanced/temporal">
    Add durability with Temporal
  </Card>

  <Card title="Examples" icon="code" href="https://github.com/lastmile-ai/mcp-agent/tree/main/examples">
    See workflows in action
  </Card>
</CardGroup>