── Native tool (single CrewAI agent) ─────────────────────────────

from pydantic import BaseModel from crewai.tools import BaseTool import requests

class CurrencyInput(BaseModel): amount: float from_currency: str to_currency: str

class CurrencyConverterTool(BaseTool): name: str = 'convert_currency' description: str = 'Convert an amount from one currency to another using live rates' args_schema: type[BaseModel] = CurrencyInput

def _run(self, amount: float, from_currency: str, to_currency: str) -> str:
    try:
        resp = requests.get(
            f'https://v6.exchangerate-api.com/v6/{API_KEY}/pair/'
            f'{from_currency}/{to_currency}/{amount}', timeout=5,
        )
        resp.raise_for_status()
        result = resp.json()['conversion_result']
        return f'{amount} {from_currency} = {result} {to_currency}'
    except Exception as e:
        return f'Conversion failed: {e}'

currency_analyst = Agent( role='Currency Analyst', goal='Provide accurate currency conversions', tools=[CurrencyConverterTool()], )

── MCP-exposed version (server.py, standalone) ───────────────────

from mcp.server.fastmcp import FastMCP

mcp = FastMCP('currency-tools')

@mcp.tool() def convert_currency(amount: float, from_currency: str, to_currency: str) -> str: '''Convert an amount from one currency to another using live rates.''' resp = requests.get( f'https://v6.exchangerate-api.com/v6/{API_KEY}/pair/' f'{from_currency}/{to_currency}/{amount}', timeout=5, ) result = resp.json()['conversion_result'] return f'{amount} {from_currency} = {result} {to_currency}'

if name == 'main': mcp.run(transport='sse') # exposes at http://localhost:8081/sse

── Consuming the MCP tool from any CrewAI agent ──────────────────

from crewai_tools import MCPServerAdapter

server_params = {'url': 'http://localhost:8081/sse'} with MCPServerAdapter(server_params) as mcp_tools: agent = Agent(role='Currency Analyst', goal='...', tools=mcp_tools)