Migrer depuis l'API des agents

agent = client.beta.agents.create(
    model="mistral-large-4",
    name="lyon-guide",
    instructions="You are a travel guide for Lyon. Answer in one short sentence.",
    completion_args={"temperature": 0.3},
)
response = client.beta.conversations.start(agent_id=agent.id, inputs="Name one museum in Lyon.")
print(response.outputs[-1].content)  # thinking and text chunks, with mistral-large-4

import asyncio

from mistralai import agents

guide = agents.Agent(
    model="mistral-large-4",
    name="lyon-guide",
    instructions="You are a travel guide for Lyon. Answer in one short sentence.",
    temperature=0.3,
)

async def create() -> None:
    result = await guide.run("Name one museum in Lyon.")
    print(result.text)

asyncio.run(create())

response = client.beta.conversations.start(
    model="mistral-large-4",
    instructions="Use your tools. Answer in one short sentence.",
    tools=[weather_tool],  # {"type": "function", "function": {"name": "get_weather", ...}}
    inputs="What is the weather in Lyon?",
)
call = next(entry for entry in response.outputs if isinstance(entry, FunctionCallEntry))
result = get_weather(**json.loads(call.arguments))
response = client.beta.conversations.append(
    conversation_id=response.conversation_id,
    inputs=[FunctionResultEntry(tool_call_id=call.tool_call_id, result=result)],
)

@agents.tool
def get_weather(city: str) -> str:
    """The weather in a city today."""
    return f"Sunny, 21 °C in {city}."

weather_guide = agents.Agent(
    model="mistral-large-4",
    instructions="Use your tools. Answer in one short sentence.",
    tools=[get_weather],
)

async def tools() -> None:
    result = await weather_guide.run("What is the weather in Lyon?")
    print(result.text)

asyncio.run(tools())

@agents.client_tool
def get_city() -> str:
    """The user's current city."""
    raise NotImplementedError  # the app that runs the agent answers it

@get_city.answer
def read_city() -> str:
    return "Lyon"  # in a real app: ask the device

located_guide = agents.Agent(
    model="mistral-large-4",
    instructions="Use your tools. Answer in one short sentence.",
    tools=[get_city],
)

async def client_tool() -> None:
    result = await located_guide.run("Which city am I in?", on_client_tool=agents.ClientToolAnswers(read_city))
    print(result.text)

asyncio.run(client_tool())

response = client.beta.conversations.start(
    model="mistral-large-4",
    instructions="Answer in one short sentence.",
    tools=[{"type": "web_search"}],
    inputs="On which days is the Musée des Confluences in Lyon closed?",
)

import os

from mistralai.client import Mistral
from mistralai.client import models as api_models

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

def text_of(content: str | list[object]) -> str:
    """The text of an Agents API message, without the model's thinking."""
    if isinstance(content, str):
        return content
    return "".join(chunk.text for chunk in content if isinstance(chunk, api_models.TextChunk))

@agents.tool
async def web_search(query: str) -> str:
    """Search the web."""
    response = await client.beta.conversations.start_async(
        model="mistral-large-4", inputs=query, tools=[{"type": "web_search"}], store=False
    )
    replies = [entry for entry in response.outputs if isinstance(entry, api_models.MessageOutputEntry)]
    return text_of(replies[-1].content)

searcher = agents.Agent(
    model="mistral-large-4",
    instructions="Use web_search for current facts. Answer in one short sentence.",
    tools=[web_search],
)

async def search() -> None:
    result = await searcher.run("On which days is the Musée des Confluences in Lyon closed?")
    print(result.text)

asyncio.run(search())

response = client.beta.conversations.start(
    model="mistral-large-4",
    instructions="Run Python to compute. Answer with the number only.",
    tools=[{"type": "code_interpreter"}],
    inputs="What is the 30th Fibonacci number?",
)

analyst = agents.Agent(
    model="mistral-large-4",
    instructions="Run Python to compute. Answer with the number only.",
    environment=agents.MistralSandbox(),
)

async def code() -> None:
    result = await analyst.run("What is the 30th Fibonacci number?")
    print(result.text)

asyncio.run(code())

response = client.beta.conversations.start(
    model="mistral-large-4",
    instructions="Answer in one short sentence.",
    tools=[{"type": "connector", "connector_id": "github_app"}],
    inputs="Which GitHub user am I?",
)

github_helper = agents.Agent(
    model="mistral-large-4",
    instructions="Answer in one short sentence.",
    connectors=[agents.connector("github_app")],
)

async def connector() -> None:
    state = await github_helper.run("Which GitHub user am I?")
    match state:
        case agents.Completed(text=text):
            print(text)
        case agents.Blocked(callbacks=callbacks):  # the user has not connected GitHub yet
            print("waiting on:", [callback.kind for callback in callbacks])
            await github_helper.session(state.session_id).cancel()

asyncio.run(connector())

response = client.beta.conversations.start(
    model="mistral-large-4",
    instructions="Answer in one short sentence.",
    tools=[{"type": "web_search", "tool_configuration": {"requires_confirmation": ["web_search"]}}],
    inputs="On which days is the Musée des Confluences in Lyon closed?",
)
pending = [entry for entry in response.outputs if isinstance(entry, FunctionCallEntry)]
response = client.beta.conversations.append(
    conversation_id=response.conversation_id,
    tool_confirmations=[{"tool_call_id": call.tool_call_id, "confirmation": "allow"} for call in pending],
)

careful_guide = agents.Agent(
    model="mistral-large-4",
    instructions="Use your tools. Answer in one short sentence.",
    tools=[get_weather],
    permission_mode="ask",  # every tool call waits for an answer
)

def approve(ctx: agents.ApprovalCallbackContext, request: agents.ApprovalRequest) -> agents.AllowApproval:
    print("approve:", request.tool.name, request.input)
    return agents.AllowApproval()

async def approvals() -> None:
    result = await careful_guide.run("What is the weather in Lyon?", on_approval=approve)
    print(result.text)

asyncio.run(approvals())

response = client.beta.conversations.start(agent_id=agent.id, inputs="Name one museum in Lyon.")
response = client.beta.conversations.append(conversation_id=response.conversation_id, inputs="When does it open?")

async def conversation() -> None:
    session = await guide.start_session()
    first = await session.run("Name one museum in Lyon.")
    print(first.text)
    second = await session.run("When does it open?")
    print(second.text)
    await session.cancel()

asyncio.run(conversation())

events = client.beta.conversations.start_stream(model="mistral-large-4", inputs="Name one museum in Lyon.")
for event in events:
    if event.event == "message.output.delta":
        print(event.data.content, end="")

def show(event: agents.Event) -> None:
    match event:
        case agents.TextDelta(text=text):
            print(text, end="", flush=True)
        case agents.TurnEnded():
            print()

async def stream() -> None:
    await guide.run("Name one museum in Lyon.", on_event=show)

asyncio.run(stream())

hotels = client.beta.agents.create(
    model="mistral-large-4",
    name="hotels",
    description="Finds a hotel with free rooms in a city.",
    instructions="Recommend the Hôtel des Célestins, rooms from 120 EUR.",
)
planner = client.beta.agents.create(
    model="mistral-large-4", name="planner", instructions="You plan trips. Hand hotel requests to the hotels agent."
)
planner = client.beta.agents.update(agent_id=planner.id, handoffs=[hotels.id])
response = client.beta.conversations.start(agent_id=planner.id, inputs="Find me a hotel in Lyon.")

@agents.tool
def find_hotel(city: str) -> str:
    """A hotel with free rooms in a city."""
    return f"Hôtel des Célestins, {city}: rooms from 120 EUR."

planner = agents.Agent(
    model="mistral-large-4",
    instructions="You plan trips. Ask the hotels agent for a hotel, then answer in one short sentence.",
    subagents=[
        agents.Subagent(
            name="hotels",
            description="Finds a hotel with free rooms in a city.",
            instructions="Call find_hotel once, then answer with the hotel and its price.",
            tools=[find_hotel],
        )
    ],
)

async def subagents() -> None:
    result = await planner.run("Find me a hotel in Lyon.")
    print(result.text)

asyncio.run(subagents())

response = client.beta.conversations.start(
    model="mistral-medium-latest",
    instructions="You are a travel guide for Lyon.",
    inputs="Name one museum in Lyon.",
    completion_args={
        "response_format": {
            "type": "json_schema",
            "json_schema": {"name": "museum", "schema": Museum.model_json_schema(), "strict": True},
        }
    },
)
museum = Museum.model_validate_json(response.outputs[-1].content)

from pydantic import BaseModel

class Museum(BaseModel):
    name: str
    district: str

typed_guide = agents.Agent(model="mistral-medium-latest", instructions="You are a travel guide for Lyon.")

async def typed() -> None:
    state = await typed_guide.run("Name one museum in Lyon.", output_type=Museum)
    match state:
        case agents.Completed(output=Museum() as museum):
            print(museum.name, "-", museum.district)
        case _:
            print(f"no museum: {state!r}")

asyncio.run(typed())

import json

def to_history(entries: list[api_models.Entry]) -> list[dict[str, object]]:
    """Turn the entries of conversations.get_history() into start_session(history=) entries."""
    history: list[dict[str, object]] = []
    for entry in entries:
        match entry:
            case api_models.MessageInputEntry() | api_models.MessageOutputEntry():
                text = text_of(entry.content)
                history.append({"type": "message", "role": entry.role, "content": [{"type": "text", "text": text}]})
            case api_models.FunctionCallEntry():
                arguments = json.loads(entry.arguments) if isinstance(entry.arguments, str) else entry.arguments
                history.append(
                    {
                        "type": "tool_call",
                        "tool_call_id": entry.tool_call_id,
                        "name": entry.name,
                        "arguments": arguments,
                    }
                )
            case api_models.FunctionResultEntry():
                history.append({"type": "tool_result", "tool_call_id": entry.tool_call_id, "result": entry.result})
    return history

async def import_conversation(conversation_id: str) -> None:
    entries = client.beta.conversations.get_history(conversation_id=conversation_id).entries
    session = await guide.start_session(history=to_history(entries))
    state = await session.run("Which museum did you name?")
    print(state.text)
    await session.cancel()

async def restart_from(conversation_id: str, entry_id: str, inputs: str) -> None:
    entries = client.beta.conversations.get_history(conversation_id=conversation_id).entries
    kept = entries[: [entry.id for entry in entries].index(entry_id) + 1]
    session = await guide.start_session(history=to_history(kept))
    state = await session.run(inputs)
    print(state.text)
    await session.cancel()