Integrations
Mistral SDK
Trace Mistral AI calls in Muster via @observe wrappers — supports streaming, async, and tool calling.
Mistral AI ships its own Python SDK (mistralai). Muster captures Mistral calls by wrapping the SDK methods with the @observe(as_type="generation") decorator. The pattern supports basic completions, chained calls, async, streaming, and tool/function calling.
Setup
%pip install mistralai langfuseimport os
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
os.environ["LANGFUSE_BASE_URL"] = "https://app.getmuster.io"
os.environ["MISTRAL_API_KEY"] = "..."Wrapping the SDK
from langfuse import observe, get_client
from mistralai import Mistral
langfuse = get_client()
mistral_client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
@observe(as_type="generation", name="Mistral-Completion")
def mistral_completion(**kwargs):
kwargs_clone = kwargs.copy()
input = kwargs_clone.pop("messages", None)
model = kwargs_clone.pop("model", None)
langfuse.update_current_generation(input=input, model=model, model_parameters=kwargs_clone)
response = mistral_client.chat.complete(model=model, messages=input, **kwargs_clone)
langfuse.update_current_generation(
output=response.choices[0].message.content,
usage_details={
"input": response.usage.prompt_tokens,
"output": response.usage.completion_tokens,
"total": response.usage.total_tokens,
},
)
return response.choices[0].message.contentPatterns covered
- Simple completion — call
mistral_completion(model=..., messages=...)from inside an@observe()-wrapped function. - Chained calls — multiple sequential calls share a single trace.
- Streaming — wrap
chat.stream(...); yield chunks and update the generation when the stream ends. - Async — wrap
chat.complete_async(...)andchat.stream_async(...). - Tool calling — log the tool selection and execution as nested observations within the generation.
Add user/session attributes
from langfuse import propagate_attributes
with propagate_attributes(user_id="user_123", session_id="sess_abc", tags=["mistral"]):
mistral_completion(...)