---
title: Decorators Reference
sidebarTitle: Decorators
description: "Author tools and workflows with MCPApp decorators."
icon: at
---

# Decorators

`MCPApp` exposes a small set of decorators that register tools, workflows, and workflow tasks. The decorators are engine-aware: when you switch from the default asyncio executor to Temporal, the same annotations automatically apply the appropriate Temporal SDK wrappers.

| Decorator | Applies to | Purpose |
| :-------- | :--------- | :------ |
| `@app.tool` | sync function | Exposes a blocking function as an MCP tool. |
| `@app.async_tool` | async function | Registers an async tool and publishes it through FastMCP. |
| `@app.workflow` | `Workflow` subclass | Declares a workflow class and adapts it for the active execution engine. |
| `@app.workflow_run` | async method | Marks the primary `run` coroutine for a workflow. |
| `@app.workflow_task` | async function/method | Registers an activity task reusable across workflows. |
| `@app.workflow_signal` | async method | Handles inbound workflow signals (Temporal-friendly). |

## Tool decorators

Tools expose code as MCP functions that agents and LLMs can call. Both decorators accept the same keyword arguments:

| Parameter | Description |
| :-------- | :---------- |
| `name` | Override the exported MCP tool name (defaults to the function name). |
| `title` | Short display label for clients. |
| `description` | Human-readable description; the function docstring is used when omitted. |
| `annotations` | Supply a `ToolAnnotations` object or mapping for MCP metadata. |
| `icons` | Provide one or more `Icon` instances or mappings for client rendering. |
| `meta` | Arbitrary metadata forwarded to FastMCP. |
| `structured_output` | Set to `True` to hint that the result is structured JSON; some LLMs will choose schema-aware models automatically. |

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

```python
from pydantic import BaseModel
from mcp_agent.app import MCPApp

app = MCPApp(name="reporting")

class Summary(BaseModel):
    title: str
    verdict: str

@app.tool(description="Summarise a raw document into a headline and verdict.", structured_output=True)
def summarise_document(text: str) -> Summary:
    """Summarise content deterministically."""
    title = text.splitlines()[0][:120]
    verdict = "APPROVED" if "ship it" in text.lower() else "NEEDS REVIEW"
    return Summary(title=title, verdict=verdict)
```

- The decorator validates the signature up-front; missing type hints or unsupported default values raise an error during import.
- `@app.tool` automatically creates a hidden workflow so the tool is reachable via both `callTool` and the workflow endpoints (`run`, `get_status`) exposed by FastMCP.
- The function executes inside the app event loop; heavy work should offload itself (for example, using `asyncio.to_thread`).

### `@app.async_tool` — asynchronous tools

```python
import httpx

@app.async_tool(
    name="fetch_page",
    description="Fetch raw HTML from an HTTP endpoint.",
    icons=[{"emoji": "🌐"}],
)
async def fetch_page(url: str, *, timeout: int = 20) -> str:
    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.text
```

- The coroutine is awaited directly, so you can call other async APIs without wrappers.
- When Temporal is active, the decorated function is wrapped with `workflow.activity` metadata automatically.

## Workflow decorator suite

Workflows orchestrate complex sequences, combining tasks, tools, and signals. Every workflow subclass must inherit from `mcp_agent.executor.workflow.Workflow`.

```python
from datetime import timedelta
from mcp_agent.executor.workflow import Workflow, WorkflowResult

@app.workflow
class PublishArticle(Workflow[WorkflowResult[str]]):
    @app.workflow_task(schedule_to_close_timeout=timedelta(minutes=5))
    async def generate_outline(self, topic: str) -> str:
        outline = f"- Introduction to {topic}\n- Key findings\n- Next steps"
        self.context.logger.info("Generated outline", data={"topic": topic})
        return outline

    @app.workflow_task(retry_policy={"maximum_attempts": 3})
    async def send_to_editor(self, outline: str) -> None:
        self.context.logger.info("Sending to editor", data={"outline": outline})

    @app.workflow_signal(name="editor_feedback")
    async def editor_feedback(self, notes: str) -> None:
        self.state.feedback = notes or ""

    @app.workflow_run
    async def run(self, topic: str) -> WorkflowResult[str]:
        outline = await self.generate_outline(topic)
        await self.send_to_editor(outline)
        feedback = getattr(self.state, "feedback", "Awaiting review.")
        return WorkflowResult(value=f"{outline}\n\nEditor feedback: {feedback}")
```

### `@app.workflow`

- Registers the class with the app and applies engine-specific decorators (`workflow.defn` for Temporal, no-op for asyncio).
- An optional `workflow_id` parameter lets you export the workflow under a different name when registering.
- The decorator stores a reference to the `MCPApp`, letting workflow instances access `self.context.app`.

### `@app.workflow_run`

Wraps the `run` coroutine so that initialization, tracing, and Temporal-specific instrumentation are handled automatically. You rarely need to call it manually—applying `@app.workflow` and naming the method `run` is enough—but explicit usage lets you decorate additional entry points.

### `@app.workflow_task`

Registers a coroutine as a reusable activity. Key options:

| Option | Effect |
| :----- | :----- |
| `name` | Override the fully qualified activity name (defaults to `<module>.<qualname>`). |
| `schedule_to_close_timeout` | Maximum wall-clock duration allowed for the task. |
| `retry_policy` | Temporal retry configuration (e.g. `{"maximum_attempts": 5}`). |
| `**meta_kwargs` | Arbitrary metadata stored alongside the task for inspection. |

The decorator enforces that the target is async; synchronous functions should wrap their blocking work with `asyncio.to_thread`.

Tasks defined outside a workflow are also supported—they are registered globally and can be reused across multiple workflows.

### `@app.workflow_signal`

Signals let external actors (humans, webhooks, other workflows) nudge a running workflow. The decorator accepts an optional `name` argument and works in both asyncio and Temporal modes.

```python
@app.workflow_signal(name="user_input")
async def collect_user_input(self, payload: str) -> None:
    self.context.logger.info("Received payload", data={"payload": payload})
    self.state.latest_payload = payload
```

The generated wrapper automatically strips the workflow instance (`self`) for Temporal’s signal handler signature.

<Info>
All workflow decorators defer to the active executor. When you switch to Temporal, tasks become activities, `run` becomes a workflow entry point, and signals map to `@workflow.signal`—no additional changes required.
</Info>

## Putting it together

```python
from mcp_agent.app import MCPApp
from mcp_agent.executor.workflow import Workflow

app = MCPApp(name="insight-generator")

@app.tool(description="List the MCP servers registered with this app.")
def list_servers() -> list[str]:
    return sorted((app.config.mcp.servers or {}).keys())

@app.workflow
class ResearchWorkflow(Workflow[None]):
    @app.workflow_task()
    async def gather_sources(self, query: str) -> list[str]:
        self.context.logger.info("Gathering sources", data={"query": query})
        return [f"https://example.com/search?q={query}"]

    @app.workflow_run
    async def run(self, topic: str) -> None:
        sources = await self.gather_sources(topic)
        self.context.logger.info("Workflow completed", data={"topic": topic, "sources": sources})
```

This pattern gives you:

- A reusable `list_servers` tool that exposes runtime metadata without boilerplate.
- A workflow that can run locally (asyncio) or durably (Temporal) with the same code.
- Optional signals/tasks to pause, resume, or branch as needed.

For CLI-driven workflows and deployment details, continue with the [CLI reference](/reference/cli). For configuration options that pair with these decorators, review the [Configuration reference](/reference/configuration).
