Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions logfire/_internal/cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -425,10 +425,11 @@ def _main(args: list[str] | None = None) -> None:
cmd_read_tokens_create.set_defaults(func=parse_create_read_token)

cmd_prompt = subparsers.add_parser('prompt', help=parse_prompt.__doc__)
agent_code_group = cmd_prompt.add_argument_group(title='code agentic specific options')
claude_or_codex_group = agent_code_group.add_mutually_exclusive_group()
claude_or_codex_group.add_argument('--claude', action='store_true', help='verify the Claude Code setup')
claude_or_codex_group.add_argument('--codex', action='store_true', help='verify the Cursor setup')
agent_code_argument_group = cmd_prompt.add_argument_group(title='code agentic specific options')
agent_code_group = agent_code_argument_group.add_mutually_exclusive_group()
agent_code_group.add_argument('--claude', action='store_true', help='verify the Claude Code setup')
agent_code_group.add_argument('--codex', action='store_true', help='verify the Cursor setup')
agent_code_group.add_argument('--opencode', action='store_true', help='verify the OpenCode setup')
cmd_prompt.add_argument('--project', action=OrgProjectAction, help='project in the format <org>/<project>')
cmd_prompt.add_argument('issue', nargs='?', help='the issue to get a prompt for')
cmd_prompt.set_defaults(func=parse_prompt)
Expand Down
99 changes: 79 additions & 20 deletions logfire/_internal/cli/prompt.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,11 +3,14 @@
from __future__ import annotations

import argparse
import json
import os
import shlex
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Any

from rich.console import Console

Expand Down Expand Up @@ -36,27 +39,11 @@ def parse_prompt(args: argparse.Namespace) -> None:
return

if args.claude:
output = subprocess.check_output(['claude', 'mcp', 'list'])
if 'logfire-mcp' not in output.decode('utf-8'):
token = _create_read_token(client, args.organization, args.project, console)
subprocess.check_output(
shlex.split(f'claude mcp add logfire -e LOGFIRE_READ_TOKEN={token} -- uvx logfire-mcp@latest')
)
console.print('Logfire MCP server added to Claude.', style='green')
configure_claude(client, args.organization, args.project, console)
elif args.codex:
codex_home = Path(os.getenv('CODEX_HOME', Path.home() / '.codex'))
codex_config = codex_home / 'config.toml'
if not codex_config.exists():
console.print('Codex config file not found. Install `codex`, or remove the `--codex` flag.')
return

codex_config_content = codex_config.read_text()

if 'logfire-mcp' not in codex_config_content:
token = _create_read_token(client, args.organization, args.project, console)
mcp_server_toml = LOGFIRE_MCP_TOML.format(token=token)
codex_config.write_text(codex_config_content + mcp_server_toml)
console.print('Logfire MCP server added to Codex.', style='green')
configure_codex(client, args.organization, args.project, console)
elif args.opencode:
configure_opencode(client, args.organization, args.project, console)

response = client.get_prompt(args.organization, args.project, args.issue)
sys.stdout.write(response['prompt'])
Expand All @@ -66,3 +53,75 @@ def _create_read_token(client: LogfireClient, organization: str, project: str, c
console.print('Logfire MCP server not found. Creating a read token...', style='yellow')
response = client.create_read_token(organization, project)
return response['token']


def configure_claude(client: LogfireClient, organization: str, project: str, console: Console) -> None:
if not shutil.which('claude'):
console.print('claude is not installed. Install `claude`, or remove the `--claude` flag.')
exit(1)

output = subprocess.check_output(['claude', 'mcp', 'list'])
if 'logfire-mcp' not in output.decode('utf-8'):
token = _create_read_token(client, organization, project, console)
subprocess.check_output(
shlex.split(f'claude mcp add logfire -e LOGFIRE_READ_TOKEN={token} -- uvx logfire-mcp@latest')
)
console.print('Logfire MCP server added to Claude.', style='green')


def configure_codex(client: LogfireClient, organization: str, project: str, console: Console) -> None:
if not shutil.which('codex'):
console.print('codex is not installed. Install `codex`, or remove the `--codex` flag.')
exit(1)

codex_home = Path(os.getenv('CODEX_HOME', Path.home() / '.codex'))
codex_config = codex_home / 'config.toml'
if not codex_config.exists():
console.print('Codex config file not found. Install `codex`, or remove the `--codex` flag.')
exit(1)

codex_config_content = codex_config.read_text()

if 'logfire-mcp' not in codex_config_content:
token = _create_read_token(client, organization, project, console)
mcp_server_toml = LOGFIRE_MCP_TOML.format(token=token)
codex_config.write_text(codex_config_content + mcp_server_toml)
console.print('Logfire MCP server added to Codex.', style='green')


def configure_opencode(client: LogfireClient, organization: str, project: str, console: Console) -> None:
# Check if opencode is installed
if not shutil.which('opencode'):
console.print('opencode is not installed. Install `opencode`, or remove the `--opencode` flag.')
exit(1)

try:
output = subprocess.check_output(['git', 'rev-parse', '--show-toplevel'])
except subprocess.CalledProcessError:
root_dir = Path.cwd()
else:
root_dir = Path(output.decode('utf-8').strip())

opencode_config = root_dir / 'opencode.jsonc'
opencode_config.touch()

opencode_config_content = opencode_config.read_text()

if 'logfire-mcp' not in opencode_config_content:
token = _create_read_token(client, organization, project, console)
if not opencode_config_content:
opencode_config.write_text(json.dumps(opencode_mcp_json(token), indent=2))
else:
opencode_config_json = json.loads(opencode_config_content)
opencode_config_json.setdefault('mcp', {})
opencode_config_json['mcp'] = {'logfire-mcp': opencode_mcp_json(token)}
opencode_config.write_text(json.dumps(opencode_config_json, indent=2))
console.print('Logfire MCP server added to OpenCode.', style='green')


def logfire_mcp_json(token: str) -> dict[str, Any]:
return {'command': 'uvx', 'args': ['logfire-mcp@latest'], 'env': {'LOGFIRE_READ_TOKEN': token}}


def opencode_mcp_json(token: str) -> dict[str, Any]:
return {'mcp': {'logfire-mcp': logfire_mcp_json(token)}}
177 changes: 173 additions & 4 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import os
import re
import shlex
import shutil
import subprocess
import sys
import types
Expand Down Expand Up @@ -1748,7 +1749,11 @@ def test_parse_prompt(prompt_http_calls: None, capsys: pytest.CaptureFixture[str
assert capsys.readouterr().out == snapshot('This is the prompt\n')


def test_parse_prompt_codex(prompt_http_calls: None, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None:
def test_parse_prompt_codex(
prompt_http_calls: None, capsys: pytest.CaptureFixture[str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(shutil, 'which', lambda x: True) # type: ignore

codex_path = tmp_path / 'codex'
codex_path.mkdir()
codex_config_path = codex_path / 'config.toml'
Expand All @@ -1772,13 +1777,28 @@ def test_parse_prompt_codex(prompt_http_calls: None, capsys: pytest.CaptureFixtu
""")


def test_parse_prompt_codex_not_installed(
prompt_http_calls: None, capsys: pytest.CaptureFixture[str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(shutil, 'which', lambda x: False) # type: ignore

with pytest.raises(SystemExit):
main(['prompt', '--project', 'fake_org/myproject', 'fix-span-issue:123', '--codex'])

assert capsys.readouterr().err == snapshot("""\
codex is not installed. Install `codex`, or remove the `--codex` flag.
""")


def test_parse_prompt_codex_config_not_found(
prompt_http_calls: None, capsys: pytest.CaptureFixture[str], tmp_path: Path
prompt_http_calls: None, capsys: pytest.CaptureFixture[str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(shutil, 'which', lambda x: True) # type: ignore

codex_path = tmp_path / 'codex'
codex_path.mkdir()

with patch.dict(os.environ, {'CODEX_HOME': str(codex_path)}):
with patch.dict(os.environ, {'CODEX_HOME': str(codex_path)}), pytest.raises(SystemExit):
main(['prompt', '--project', 'fake_org/myproject', 'fix-span-issue:123', '--codex'])

assert capsys.readouterr().err == snapshot(
Expand All @@ -1787,8 +1807,10 @@ def test_parse_prompt_codex_config_not_found(


def test_parse_prompt_codex_logfire_mcp_installed(
prompt_http_calls: None, capsys: pytest.CaptureFixture[str], tmp_path: Path
prompt_http_calls: None, capsys: pytest.CaptureFixture[str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(shutil, 'which', lambda x: True) # type: ignore

codex_path = tmp_path / 'codex'
codex_path.mkdir()
codex_config_path = codex_path / 'config.toml'
Expand All @@ -1803,6 +1825,8 @@ def test_parse_prompt_codex_logfire_mcp_installed(
def test_parse_prompt_claude(
prompt_http_calls: None, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(shutil, 'which', lambda x: True) # type: ignore

def logfire_mcp_installed(_: list[str]) -> bytes:
return b'logfire-mcp is installed'

Expand All @@ -1812,9 +1836,24 @@ def logfire_mcp_installed(_: list[str]) -> bytes:
assert capsys.readouterr().out == snapshot('This is the prompt\n')


def test_parse_prompt_claude_not_installed(
prompt_http_calls: None, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(shutil, 'which', lambda x: False) # type: ignore

with pytest.raises(SystemExit):
main(['prompt', '--project', 'fake_org/myproject', 'fix-span-issue:123', '--claude'])

assert capsys.readouterr().err == snapshot("""\
claude is not installed. Install `claude`, or remove the `--claude` flag.
""")


def test_parse_prompt_claude_no_mcp(
prompt_http_calls: None, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(shutil, 'which', lambda x: True) # type: ignore

def logfire_mcp_installed(_: list[str]) -> bytes:
return b'not installed'

Expand All @@ -1829,6 +1868,136 @@ def logfire_mcp_installed(_: list[str]) -> bytes:
""")


def test_parse_prompt_opencode(
prompt_http_calls: None,
capsys: pytest.CaptureFixture[str],
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(shutil, 'which', lambda x: True) # type: ignore
monkeypatch.setattr(Path, 'cwd', lambda: tmp_path)

def check_output(x: list[str]) -> bytes:
return tmp_path.as_posix().encode('utf-8')

monkeypatch.setattr(subprocess, 'check_output', check_output)

main(['prompt', '--project', 'fake_org/myproject', 'fix-span-issue:123', '--opencode'])

out, err = capsys.readouterr()
assert out == snapshot("""\
This is the prompt
""")
assert err == snapshot("""\
Logfire MCP server not found. Creating a read token...
Logfire MCP server added to OpenCode.
""")


def test_parse_prompt_opencode_no_git(
prompt_http_calls: None,
capsys: pytest.CaptureFixture[str],
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(shutil, 'which', lambda x: True) # type: ignore
monkeypatch.setattr(Path, 'cwd', lambda: tmp_path)

def check_output(x: list[str]) -> bytes:
raise subprocess.CalledProcessError(1, x)

monkeypatch.setattr(subprocess, 'check_output', check_output)

main(['prompt', '--project', 'fake_org/myproject', 'fix-span-issue:123', '--opencode'])

out, err = capsys.readouterr()
assert out == snapshot("""\
This is the prompt
""")
assert err == snapshot("""\
Logfire MCP server not found. Creating a read token...
Logfire MCP server added to OpenCode.
""")


def test_parse_prompt_opencode_not_installed(
prompt_http_calls: None,
capsys: pytest.CaptureFixture[str],
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(shutil, 'which', lambda x: False) # type: ignore
monkeypatch.setattr(Path, 'cwd', lambda: tmp_path)

with pytest.raises(SystemExit):
main(['prompt', '--project', 'fake_org/myproject', 'fix-span-issue:123', '--opencode'])

out, err = capsys.readouterr()
assert out == snapshot('')
assert err == snapshot("""\
opencode is not installed. Install `opencode`, or remove the `--opencode` flag.
""")


def test_parse_prompt_opencode_logfire_mcp_installed(
prompt_http_calls: None,
capsys: pytest.CaptureFixture[str],
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(shutil, 'which', lambda x: True) # type: ignore
monkeypatch.setattr(Path, 'cwd', lambda: tmp_path)

(tmp_path / 'opencode.jsonc').write_text("""
{
"mcp": {
"logfire-mcp": {
"command": "uvx",
"args": ["logfire-mcp@latest"],
"env": {"LOGFIRE_READ_TOKEN": "fake_token"}
}
}
}
""")

def check_output(x: list[str]) -> bytes:
return tmp_path.as_posix().encode('utf-8')

monkeypatch.setattr(subprocess, 'check_output', check_output)

main(['prompt', '--project', 'fake_org/myproject', 'fix-span-issue:123', '--opencode'])

out, err = capsys.readouterr()
assert out == snapshot('This is the prompt\n')
assert err == snapshot('')


def test_parse_opencode_logfire_mcp_not_installed_with_existing_config(
prompt_http_calls: None,
capsys: pytest.CaptureFixture[str],
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(shutil, 'which', lambda x: True) # type: ignore
monkeypatch.setattr(Path, 'cwd', lambda: tmp_path)

(tmp_path / 'opencode.jsonc').write_text('{}')

def check_output(x: list[str]) -> bytes:
return tmp_path.as_posix().encode('utf-8')

monkeypatch.setattr(subprocess, 'check_output', check_output)

main(['prompt', '--project', 'fake_org/myproject', 'fix-span-issue:123', '--opencode'])

out, err = capsys.readouterr()
assert out == snapshot('This is the prompt\n')
assert err == snapshot("""\
Logfire MCP server not found. Creating a read token...
Logfire MCP server added to OpenCode.
""")


def test_base_url_and_logfire_url(
tmp_dir_cwd: Path, logfire_credentials: LogfireCredentials, capsys: pytest.CaptureFixture[str]
):
Expand Down
Loading