mcp elicitation

[pgrayy]

Description

Allow customers to pass an elicitation callback to the MCPClient (docs).

Note, I did prototype running elicitation through the Strands interrupt workflow but decided to hold off for the following reasons:

1. Unlike interrupts, MCP elicitation requires a persistent connection to the tool server. If the MCPClient is torn down, the elicitation request is lost.
2. Because the connection has to remain open at all times, I figured it would be easier and more intuitive for customers to handle the elicitation requests in the async callback as designed. They don't have to parse the interrupts from AgentResult and pass in a separate interruptResponses payload. Everything can just be handled in the callback.
3. Users cannot raise an interrupt during a direct tool call (details). An elicitation callback will run fine though.
4. We always have the option to run elicitation through the interrupt system in the future. We could check if the elicitation_callback is None and if so, we provide our own callback that translates the elicitation requests into InterruptException's.

Usage

# **server.py**
from mcp.server import FastMCP
from mcp.types import ElicitRequest, ElicitRequestParams, ElicitResult

server = FastMCP("mytools")


@server.tool()
async def delete_file(paths: list[str]) -> str:
    request = ElicitRequest(
        params=ElicitRequestParams(
            message=f"Do you want to delete {paths}",
            requestedSchema={
                "type": "object",
                "properties": {
                    "username": {"type": "string", "description": "Who is approving?"},
                },
                "required": ["username"]
            }
        )
    )
    result = await server.get_context().session.send_request(request, ElicitResult)

    action = result.action
    username = result.content["username"]

    if action != "accept":
        return f"User {username} rejected deletion"

    return f"User {username} approved deletion"
    

server.run()

# **client.py**
import json

from mcp import stdio_client, StdioServerParameters
from mcp.types import ElicitResult

from strands import Agent
from strands.tools.mcp import MCPClient


async def elicitation_callback(context, params):
    print(f"ELICITATION: {params.message}")
    
    return ElicitResult(
        action="accept",  # or try "decline"
        content={"username": "pgrayy"}
    )


client = MCPClient(
    lambda: stdio_client(
        StdioServerParameters(command="python", args=["/path/to/server.py"])
    ),
    elicitation_callback=elicitation_callback,
)
with client:
    agent = Agent(tools=client.list_tools_sync(), callback_handler=None)

    result = agent("Delete 'a/b/c.txt' and 'd/e/f.txt' and share the name of the approver")
    print(f"MESSAGE: {json.dumps(result.message)}")

Related Issues

#204

Documentation PR

strands-agents/docs#300

Type of Change

New feature

Testing

How have you tested the change? Verify that the changes do not break functionality or introduce warnings in consuming repositories: agents-docs, agents-tools, agents-cli

• I ran hatch run prepare
• I ran hatch test tests_integ/mcp/test_mcp_elicitation.py: Wrote new integ tests.

Checklist

• I have read the CONTRIBUTING document
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
• My changes generate no new warnings
• Any dependent changes have been merged and published

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!