Skip to content

CLI Package

The lampe.cli package provides command-line interface functionality for the SDK.

Commands

lampe.cli.commands

check_reviewed

check_reviewed(repo: Path = typer.Option(..., exists=True, file_okay=False, dir_okay=True, readable=True), repo_full_name: str | None = typer.Option(None, help='Repository full name (e.g. owner/repo)'), output: str = typer.Option('auto', help='Output provider (auto|console|github|gitlab|bitbucket)'), pr_number: int | None = typer.Option(None, '--pr', help='Pull request number (required for non-console providers)'))

Check if the token user has already reviewed this PR.

Returns exit code 0 if reviewed, 1 if not reviewed.

Source code in packages/lampe-cli/src/lampe/cli/commands/check_reviewed.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def check_reviewed(
    repo: Path = typer.Option(..., exists=True, file_okay=False, dir_okay=True, readable=True),
    repo_full_name: str | None = typer.Option(None, help="Repository full name (e.g. owner/repo)"),
    output: str = typer.Option("auto", help="Output provider (auto|console|github|gitlab|bitbucket)"),
    pr_number: int | None = typer.Option(None, "--pr", help="Pull request number (required for non-console providers)"),
):
    """Check if the token user has already reviewed this PR.

    Returns exit code 0 if reviewed, 1 if not reviewed.
    """
    initialize()
    repo_model = Repository(local_path=str(repo), full_name=repo_full_name)
    pr_model = PullRequest(
        number=pr_number or 0,
        title="",
        body=None,
        base_commit_hash="",
        base_branch_name="",
        head_commit_hash="",
        head_branch_name="",
    )

    try:
        provider = Provider.create_provider(provider_name=output, repository=repo_model, pull_request=pr_model)
    except ValueError as e:
        if "required" in str(e).lower() and "pr" in str(e).lower():
            print(f"❌ Error: PR number is required for {output} provider. Use --pr <number>", file=sys.stderr)
            sys.exit(1)
        raise

    try:
        has_reviewed = provider.has_reviewed()
        if has_reviewed:
            print("✅ PR has already been reviewed by the token user")
            sys.exit(0)
        else:
            print("❌ PR has not been reviewed by the token user yet")
            sys.exit(1)
    except Exception as e:
        print(f"❌ Error checking if PR has been reviewed: {e}", file=sys.stderr)
        sys.exit(1)

describe

describe(repo: Path = typer.Option(..., exists=True, file_okay=False, dir_okay=True, readable=True), repo_full_name: str | None = typer.Option(None, help='Repository full name (e.g. owner/repo)'), base: str = typer.Option(..., help='Base commit SHA'), head: str = typer.Option(..., help='Head commit SHA'), title: str = typer.Option('Pull Request', help='PR title (local runs)'), output: str = typer.Option('auto', help='Output provider (auto|console|github|gitlab|bitbucket)'), variant: str = typer.Option('default', help='default|agentic'), files_exclude: list[str] | None = typer.Option(None, '--exclude'), files_reinclude: list[str] | None = typer.Option(None, '--reinclude'), truncation_tokens: int = typer.Option(DEFAULT_MAX_TOKENS, '--max-tokens'), timeout: int | None = typer.Option(None, '--timeout-seconds'), verbose: bool = typer.Option(False, '--verbose/--no-verbose'))

Generate a PR description and deliver it to the specified output provider.

Source code in packages/lampe-cli/src/lampe/cli/commands/describe.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def describe(
    repo: Path = typer.Option(..., exists=True, file_okay=False, dir_okay=True, readable=True),
    repo_full_name: str | None = typer.Option(None, help="Repository full name (e.g. owner/repo)"),
    base: str = typer.Option(..., help="Base commit SHA"),
    head: str = typer.Option(..., help="Head commit SHA"),
    title: str = typer.Option("Pull Request", help="PR title (local runs)"),
    output: str = typer.Option("auto", help="Output provider (auto|console|github|gitlab|bitbucket)"),
    variant: str = typer.Option("default", help="default|agentic"),
    files_exclude: list[str] | None = typer.Option(None, "--exclude"),
    files_reinclude: list[str] | None = typer.Option(None, "--reinclude"),
    truncation_tokens: int = typer.Option(DEFAULT_MAX_TOKENS, "--max-tokens"),
    timeout: int | None = typer.Option(None, "--timeout-seconds"),
    verbose: bool = typer.Option(False, "--verbose/--no-verbose"),
):
    """Generate a PR description and deliver it to the specified output provider."""
    initialize()
    repo_model = Repository(local_path=str(repo), full_name=repo_full_name)
    pr_model = PullRequest(
        number=0,
        title=title,
        body=None,
        base_commit_hash=base,
        base_branch_name="",
        head_commit_hash=head,
        head_branch_name="",
    )

    provider = Provider.create_provider(provider_name=output, repository=repo_model, pull_request=pr_model)

    generator = DefaultGeneratorAdapter() if variant == "default" else AgenticGeneratorAdapter()
    pr_cfg = PRDescriptionConfig(
        files_exclude_patterns=list(files_exclude) if files_exclude else None,
        files_reinclude_patterns=list(files_reinclude) if files_reinclude else None,
        truncation_tokens=truncation_tokens,
        timeout=timeout,
        verbose=verbose,
    )

    async def _run():
        workflow_task = PRDescriptionOrchestratorWorkflow(
            provider=provider, generator=generator, timeout=timeout, verbose=verbose
        )
        await workflow_task.run(
            start_event=PRDescriptionStart(repository=repo_model, pull_request=pr_model, config=pr_cfg)
        )

    asyncio.run(_run())

healthcheck

healthcheck() -> None

Check if the CLI is healthy and can connect to the configured provider.

Source code in packages/lampe-cli/src/lampe/cli/commands/healthcheck.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def healthcheck() -> None:
    """Check if the CLI is healthy and can connect to the configured provider."""
    logger.info("🔍 Checking CLI health...")
    initialize()
    # Create dummy repository and pull request objects for testing
    repo = Repository(local_path=".", full_name="test/repo")
    pr = PullRequest(
        number=1,
        title="Test PR",
        base_commit_hash="test-base",
        base_branch_name="main",
        head_commit_hash="test-head",
        head_branch_name="feature/test",
    )

    # Initialize provider and run healthcheck
    try:
        provider: Provider = Provider.create_provider("auto", repository=repo, pull_request=pr)
        provider.healthcheck()

        # Check LLM API keys
        logger.info("🔑 Checking LLM API keys...")
        openai_key = os.getenv("OPENAI_API_KEY")
        anthropic_key = os.getenv("ANTHROPIC_API_KEY")

        if not openai_key and not anthropic_key:
            logger.info("❌ No LLM API keys found")
            logger.info("   Set at least one of:")
            logger.info("   - OPENAI_API_KEY for OpenAI models")
            logger.info("   - ANTHROPIC_API_KEY for Anthropic models")
            sys.exit(1)

        if openai_key:
            logger.info("✅ OPENAI_API_KEY is set")
        if anthropic_key:
            logger.info("✅ ANTHROPIC_API_KEY is set")

        logger.info("\n🎉 All health checks passed! CLI is ready to use.")

    except Exception as e:
        logger.exception(f"❌ Health check failed: {e}")
        sys.exit(1)

review

review(repo: Path = typer.Option(..., exists=True, file_okay=False, dir_okay=True, readable=True), repo_full_name: str | None = typer.Option(None, help='Repository full name (e.g. owner/repo)'), base: str = typer.Option(..., help='Base commit SHA'), head: str = typer.Option(..., help='Head commit SHA'), title: str = typer.Option('Pull Request', help='PR title (local runs)'), output: str = typer.Option('auto', help='Output provider (auto|console|github|gitlab|bitbucket)'), review_depth: ReviewDepth = typer.Option(ReviewDepth.STANDARD, help='Review depth (basic|standard|comprehensive)'), variant: str = typer.Option('multi-agent', help='Review variant (multi-agent|diff-by-diff)'), guidelines: list[str] | None = typer.Option(None, '--guideline', help='Custom review guidelines (can be repeated)'), files_exclude: list[str] | None = typer.Option(None, '--exclude'), timeout: int | None = typer.Option(None, '--timeout-seconds'), verbose: bool = typer.Option(False, '--verbose/--no-verbose'))

Generate a PR code review and deliver it to the specified output provider.

Model selection is automatic based on review_depth: - basic: gpt-5-nano - standard: gpt-5 - comprehensive: gpt-5.1

Source code in packages/lampe-cli/src/lampe/cli/commands/review.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def review(
    repo: Path = typer.Option(..., exists=True, file_okay=False, dir_okay=True, readable=True),
    repo_full_name: str | None = typer.Option(None, help="Repository full name (e.g. owner/repo)"),
    base: str = typer.Option(..., help="Base commit SHA"),
    head: str = typer.Option(..., help="Head commit SHA"),
    title: str = typer.Option("Pull Request", help="PR title (local runs)"),
    output: str = typer.Option("auto", help="Output provider (auto|console|github|gitlab|bitbucket)"),
    review_depth: ReviewDepth = typer.Option(ReviewDepth.STANDARD, help="Review depth (basic|standard|comprehensive)"),
    variant: str = typer.Option("multi-agent", help="Review variant (multi-agent|diff-by-diff)"),
    guidelines: list[str] | None = typer.Option(None, "--guideline", help="Custom review guidelines (can be repeated)"),
    files_exclude: list[str] | None = typer.Option(None, "--exclude"),
    timeout: int | None = typer.Option(None, "--timeout-seconds"),
    verbose: bool = typer.Option(False, "--verbose/--no-verbose"),
):
    """Generate a PR code review and deliver it to the specified output provider.

    Model selection is automatic based on review_depth:
    - basic: gpt-5-nano
    - standard: gpt-5
    - comprehensive: gpt-5.1
    """
    initialize()
    repo_model = Repository(local_path=str(repo), full_name=repo_full_name)
    pr_model = PullRequest(
        number=0,
        title=title,
        body=None,
        base_commit_hash=base,
        base_branch_name="",
        head_commit_hash=head,
        head_branch_name="",
    )

    provider = Provider.create_provider(provider_name=output, repository=repo_model, pull_request=pr_model)

    generator = DiffByDiffReviewAdapter() if variant == "diff-by-diff" else AgenticReviewAdapter()
    pr_cfg = PRReviewConfig(
        review_depth=review_depth,
        custom_guidelines=guidelines,
        files_exclude_patterns=files_exclude,
        agents_required=[DefaultAgent],
        timeout=timeout,
        verbose=verbose,
    )

    async def _run():
        workflow_task = PRReviewOrchestratorWorkflow(
            provider=provider, generator=generator, timeout=timeout, verbose=verbose
        )
        await workflow_task.run(start_event=PRReviewStart(repository=repo_model, pull_request=pr_model, config=pr_cfg))

    asyncio.run(_run())

Orchestrators

lampe.cli.orchestrators

Providers

lampe.cli.providers

base

Provider(repository: Repository, pull_request: PullRequest)

Bases: ABC

Abstract provider for delivering workflow outputs.

Source code in packages/lampe-cli/src/lampe/cli/providers/base.py 
78
79
80
def __init__(self, repository: Repository, pull_request: PullRequest) -> None:
    self.repository = repository
    self.pull_request = pull_request
create_provider(provider_name: ProviderType | str, repository: Repository, pull_request: PullRequest) -> 'Provider' staticmethod

Create a provider instance based on the specified type.

Source code in packages/lampe-cli/src/lampe/cli/providers/base.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@staticmethod
def create_provider(
    provider_name: ProviderType | str, repository: Repository, pull_request: PullRequest
) -> "Provider":
    """Create a provider instance based on the specified type."""
    if isinstance(provider_name, str):
        # Handle "auto" detection
        if provider_name == "auto":
            provider_name = Provider.detect_provider_type()
        else:
            provider_name = ProviderType(provider_name)

    if provider_name == ProviderType.CONSOLE:
        from lampe.cli.providers.console import ConsoleProvider

        return ConsoleProvider(repository=repository, pull_request=pull_request)
    elif provider_name == ProviderType.GITHUB:
        from lampe.cli.providers.github import GitHubProvider

        return GitHubProvider(repository=repository, pull_request=pull_request)
    elif provider_name == ProviderType.BITBUCKET:
        from lampe.cli.providers.bitbucket import BitbucketProvider

        return BitbucketProvider(repository=repository, pull_request=pull_request)
    else:
        raise ValueError(f"Provider type {provider_name} not yet implemented")
deliver_pr_description(payload: PRDescriptionPayload) -> None abstractmethod

Deliver a PR description to the configured destination.

Source code in packages/lampe-cli/src/lampe/cli/providers/base.py 
82
83
84
85
@abstractmethod
def deliver_pr_description(self, payload: PRDescriptionPayload) -> None:
    """Deliver a PR description to the configured destination."""
    ...
deliver_pr_review(payload: PRReviewPayload) -> None abstractmethod

Deliver a PR review to the configured destination.

Source code in packages/lampe-cli/src/lampe/cli/providers/base.py 
87
88
89
90
@abstractmethod
def deliver_pr_review(self, payload: PRReviewPayload) -> None:
    """Deliver a PR review to the configured destination."""
    ...
detect_provider_type() -> ProviderType staticmethod

Detect the appropriate provider type based on available environment variables.

Source code in packages/lampe-cli/src/lampe/cli/providers/base.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@staticmethod
def detect_provider_type() -> ProviderType:
    """Detect the appropriate provider type based on available environment variables."""
    # Priority order for provider detection
    env_var_mapping = {
        "GITHUB_API_TOKEN": ProviderType.GITHUB,
        "GITHUB_TOKEN": ProviderType.GITHUB,
        "LAMPE_GITHUB_TOKEN": ProviderType.GITHUB,
        "LAMPE_GITHUB_APP_ID": ProviderType.GITHUB,
        "LAMPE_GITHUB_APP_PRIVATE_KEY": ProviderType.GITHUB,
        "GITLAB_API_TOKEN": ProviderType.GITLAB,
        "LAMPE_BITBUCKET_TOKEN": ProviderType.BITBUCKET,
        "LAMPE_BITBUCKET_APP_KEY": ProviderType.BITBUCKET,
        "BITBUCKET_WORKSPACE": ProviderType.BITBUCKET,
    }

    for env_var, provider_type in env_var_mapping.items():
        if os.getenv(env_var):
            return provider_type

    # Fallback to console if no API tokens are found
    return ProviderType.CONSOLE
has_reviewed() -> bool abstractmethod

Check if the token user has already reviewed this PR.

Source code in packages/lampe-cli/src/lampe/cli/providers/base.py 
 97
 98
 99
100
@abstractmethod
def has_reviewed(self) -> bool:
    """Check if the token user has already reviewed this PR."""
    ...
healthcheck() -> None abstractmethod

Check if the provider is healthy and can connect to the service.

Source code in packages/lampe-cli/src/lampe/cli/providers/base.py 
92
93
94
95
@abstractmethod
def healthcheck(self) -> None:
    """Check if the provider is healthy and can connect to the service."""
    ...

ProviderType

Bases: StrEnum

Available provider types.

update_or_add_text_between_tags(text: str, new_text: str, feature: str) -> str

Update the text between the tags and with new_text. If the tags don't exist, add them at the bottom of the text. The tags and new_text are preserved in the output.

Source code in packages/lampe-cli/src/lampe/cli/providers/base.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def update_or_add_text_between_tags(text: str, new_text: str, feature: str) -> str:
    """
    Update the text between the tags [](lampe-sdk-{feature}-start) and [](lampe-sdk-{feature}-end)
    with new_text. If the tags don't exist, add them at the bottom of the text.
    The tags and new_text are preserved in the output.
    """
    identifier = f"lampe-sdk-{feature}-start"
    start_tag = rf"\[\]\(lampe-sdk-{feature}-start\)"
    end_tag = rf"\[\]\(lampe-sdk-{feature}-end\)"

    pattern = re.compile(rf"({start_tag})(.*?|\s*?){end_tag}", re.DOTALL)

    def replacer(match):
        return f"{match.group(1)}\n{new_text}\n[]({identifier.replace('-start', '')}-end)"

    # Try to replace the first occurrence
    result, count = pattern.subn(replacer, text, count=1)

    # If no tags were found, add them at the bottom
    if count == 0:
        result = f"{text}\n\n[]({identifier})\n{new_text}\n[]({identifier.replace('-start', '')}-end)"

    return result

bitbucket

BitbucketProvider(repository: Repository, pull_request: PullRequest)

Bases: Provider

Bitbucket provider for delivering PR descriptions to Bitbucket Cloud API.

Source code in packages/lampe-cli/src/lampe/cli/providers/bitbucket.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def __init__(self, repository: Repository, pull_request: PullRequest) -> None:
    if pull_request.number == 0:
        # Try Bitbucket Pipelines environment variable first, then fallback to PR_NUMBER
        pr_number = os.getenv("BITBUCKET_PR_ID") or os.getenv("PR_NUMBER")
        if not pr_number:
            raise ValueError("BITBUCKET_PR_ID or PR_NUMBER environment variable is required for Bitbucket provider")
        pull_request.number = int(pr_number)

    super().__init__(repository, pull_request)

    # Extract workspace and repository from environment variables
    self.workspace = os.getenv("BITBUCKET_WORKSPACE")
    self.repo_slug = os.getenv("BITBUCKET_REPO_SLUG")

    if not self.workspace or not self.repo_slug:
        raise ValueError(
            "BITBUCKET_WORKSPACE and BITBUCKET_REPO_SLUG environment variables are required for Bitbucket provider"
        )

    # Initialize Bitbucket client with appropriate authentication
    self.base_url, self.auth_headers = self._initialize_bitbucket_client()
deliver_pr_description(payload: PRDescriptionPayload) -> None

Update the PR description on Bitbucket.

Source code in packages/lampe-cli/src/lampe/cli/providers/bitbucket.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def deliver_pr_description(self, payload: PRDescriptionPayload) -> None:
    """Update the PR description on Bitbucket."""
    if self.pull_request.number == 0:
        raise ValueError("Cannot update Bitbucket PR description for local run")

    try:
        # Get current PR details
        pr_url = (
            f"{self.base_url}/2.0/repositories/{self.workspace}/"
            f"{self.repo_slug}/pullrequests/{self.pull_request.number}"
        )

        # Fetch current PR to get existing description
        response = requests.get(pr_url, headers=self.auth_headers)
        response.raise_for_status()
        pr_data = response.json()

        # Update description with new content
        current_description = pr_data.get("description", "") or ""
        new_description = update_or_add_text_between_tags(
            current_description, payload.description_with_title, "description"
        )

        # Update the PR
        update_data = {"description": new_description}
        update_response = requests.put(pr_url, json=update_data, headers=self.auth_headers)
        update_response.raise_for_status()

        logger.info(f"✅ Successfully updated PR #{self.pull_request.number} description on Bitbucket")
    except requests.exceptions.RequestException as e:
        logger.error(f"❌ Failed to update Bitbucket PR: {e}")
        # Fallback to console output
        logger.info("Description:")
        logger.info(payload.description)
    except Exception as e:
        logger.error(f"❌ Unexpected error updating Bitbucket PR: {e}")
        # Fallback to console output
        logger.info("Description:")
        logger.info(payload.description)
deliver_pr_review(payload: PRReviewPayload) -> None

Post PR review comments on Bitbucket.

Source code in packages/lampe-cli/src/lampe/cli/providers/bitbucket.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def deliver_pr_review(self, payload: PRReviewPayload) -> None:
    """Post PR review comments on Bitbucket."""
    if self.pull_request.number == 0:
        raise ValueError("Cannot post Bitbucket PR review for local run")

    try:
        # Post review comments for each agent review
        for agent_review in payload.reviews:
            # Post agent summary comment
            if agent_review.summary:
                try:
                    comment_url = (
                        f"{self.base_url}/2.0/repositories/{self.workspace}/"
                        f"{self.repo_slug}/pullrequests/{self.pull_request.number}/comments"
                    )
                    comment_data = {
                        "content": {
                            "raw": f"## {agent_review.agent_name}\n\n"
                            f"**Focus Areas:** {', '.join(agent_review.focus_areas)}\n\n"
                            f"{agent_review.summary}"
                        }
                    }
                    response = requests.post(comment_url, json=comment_data, headers=self.auth_headers)
                    response.raise_for_status()
                except Exception as e:
                    logger.warning(f"Failed to post agent summary for {agent_review.agent_name}: {e}")

            # Post file-specific comments
            for file_review in agent_review.reviews:
                if file_review.line_comments:
                    # Create review comments for specific lines
                    for line, comment in file_review.line_comments.items():
                        try:
                            line_number = int(line)
                        except ValueError:
                            match = re.match(r"\D*(\d+)", str(line))
                            if match:
                                line_number = int(match.group(1))
                            else:
                                line_number = 0
                        try:
                            # Post a comment on the PR
                            comment_url = (
                                f"{self.base_url}/2.0/repositories/{self.workspace}/"
                                f"{self.repo_slug}/pullrequests/{self.pull_request.number}/comments"
                            )
                            comment_data = {
                                "content": {"raw": f"## 🔦🐛\n{comment}"},
                                "inline": {
                                    "from": line_number - 1 if line_number != 0 else 0,
                                    "to": line_number,
                                    "start_from": line_number - 1 if line_number != 0 else 0,
                                    "start_to": line_number,
                                    "path": file_review.file_path,
                                },
                            }
                            response = requests.post(comment_url, json=comment_data, headers=self.auth_headers)
                            response.raise_for_status()
                        except Exception as e:
                            logger.warning(f"Failed to post comment for {file_review.file_path}:{line}: {e}")

                # Post file summary comment if no line comments
                if not file_review.line_comments and file_review.summary:
                    try:
                        comment_url = (
                            f"{self.base_url}/2.0/repositories/{self.workspace}/"
                            f"{self.repo_slug}/pullrequests/{self.pull_request.number}/comments"
                        )
                        comment_data = {"content": {"raw": f"**{file_review.file_path}:** {file_review.summary}"}}
                        response = requests.post(comment_url, json=comment_data, headers=self.auth_headers)
                        response.raise_for_status()
                    except Exception as e:
                        logger.warning(f"Failed to post summary for {file_review.file_path}: {e}")

        logger.info(f"✅ Successfully posted PR #{self.pull_request.number} review comments on Bitbucket")
    except requests.exceptions.RequestException as e:
        logger.error(f"❌ Failed to post Bitbucket PR review: {e}")
        # Fallback to console output
        logger.info("Review:")
        logger.info(payload.review_markdown)
    except Exception as e:
        logger.error(f"❌ Unexpected error posting Bitbucket PR review: {e}")
        # Fallback to console output
        logger.info("Review:")
        logger.info(payload.review_markdown)
has_reviewed() -> bool

Check if the token user has already reviewed this PR.

Source code in packages/lampe-cli/src/lampe/cli/providers/bitbucket.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
def has_reviewed(self) -> bool:
    """Check if the token user has already reviewed this PR."""
    if self.pull_request.number == 0:
        return False

    try:
        # Get PR comments
        comments_url = (
            f"{self.base_url}/2.0/repositories/{self.workspace}/"
            f"{self.repo_slug}/pullrequests/{self.pull_request.number}/comments"
        )
        comments_response = requests.get(comments_url, headers=self.auth_headers)
        comments_response.raise_for_status()
        comments_data = comments_response.json()

        # Try to get the current authenticated user (token owner)
        # This works for user tokens but may fail for repository/workspace tokens
        token_user_uuid = None
        token_username = None
        try:
            user_info_response = requests.get(f"{self.base_url}/2.0/user", headers=self.auth_headers)
            user_info_response.raise_for_status()
            user_info = user_info_response.json()
            token_user_uuid = user_info.get("uuid") or user_info.get("account_id")
            if not token_user_uuid:
                token_username = user_info.get("username") or user_info.get("nickname")
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 401:
                # Repository/workspace tokens can't access /2.0/user
                # Fall back to pattern-based detection
                logger.debug("Token doesn't have access to /2.0/user endpoint, using pattern-based detection")
            else:
                raise

        # Check for comments by the token user (if we have user identity)
        if token_user_uuid or token_username:
            for comment in comments_data.get("values", []):
                user = comment.get("user", {})
                if token_user_uuid:
                    if user.get("uuid") == token_user_uuid or user.get("account_id") == token_user_uuid:
                        return True
                elif token_username:
                    if user.get("username") == token_username or user.get("nickname") == token_username:
                        return True

        # Fallback: Check for review comments by pattern (for repository/workspace tokens)
        # Look for comments that match Lampe review format:
        # - Comments starting with "## " (agent name headers)
        # - Comments containing "Focus Areas:"
        # - Comments containing "🔦🐛" (line comment marker)
        review_patterns = [
            r"^##\s+\w+",  # Agent name header (e.g., "## SecurityAgent")
            r"\*\*Focus Areas:\*\*",  # Focus areas marker
            r"##\s*🔦🐛",  # Line comment marker
        ]

        for comment in comments_data.get("values", []):
            content = comment.get("content", {}).get("raw", "") or comment.get("content", {}).get("markup", "")
            if content:
                for pattern in review_patterns:
                    if re.search(pattern, content, re.IGNORECASE | re.MULTILINE):
                        return True

        return False
    except requests.exceptions.RequestException as e:
        logger.warning(f"Failed to check if PR has been reviewed: {e}")
        return False
    except Exception as e:
        logger.warning(f"Unexpected error checking if PR has been reviewed: {e}")
        return False
healthcheck() -> None

Check if the Bitbucket provider is healthy and can connect to Bitbucket.

Source code in packages/lampe-cli/src/lampe/cli/providers/bitbucket.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def healthcheck(self) -> None:
    """Check if the Bitbucket provider is healthy and can connect to Bitbucket."""
    logger.info("🔍 Checking Bitbucket provider health...")

    # Check Bitbucket environment variables
    workspace = os.getenv("BITBUCKET_WORKSPACE")
    repo_slug = os.getenv("BITBUCKET_REPO_SLUG")

    if not workspace or not repo_slug:
        logger.info("❌ Bitbucket environment variables not set")
        logger.info("   Set both:")
        logger.info("   - BITBUCKET_WORKSPACE (e.g., 'my-workspace')")
        logger.info("   - BITBUCKET_REPO_SLUG (e.g., 'my-repo')")
        raise ValueError("BITBUCKET_WORKSPACE and BITBUCKET_REPO_SLUG environment variables are required")

    logger.info(f"✅ BITBUCKET_WORKSPACE set to: {workspace}")
    logger.info(f"✅ BITBUCKET_REPO_SLUG set to: {repo_slug}")

    # Check authentication environment variables
    token = os.getenv("LAMPE_BITBUCKET_TOKEN")
    app_key = os.getenv("LAMPE_BITBUCKET_APP_KEY")
    app_secret = os.getenv("LAMPE_BITBUCKET_APP_SECRET")

    auth_method = None
    if token:
        auth_method = "Token"
        logger.info("✅ Bitbucket token authentication detected")
    elif app_key and app_secret:
        auth_method = "App"
        logger.info("✅ Bitbucket App authentication detected")
    else:
        logger.info("❌ No Bitbucket authentication found")
        logger.info("   Set either:")
        logger.info("   - LAMPE_BITBUCKET_TOKEN for token authentication")
        logger.info("   - LAMPE_BITBUCKET_APP_KEY and LAMPE_BITBUCKET_APP_SECRET for app authentication")
        raise ValueError("No Bitbucket authentication found")

    # Test Bitbucket connection
    try:
        # Test API access by getting repository info
        repo_url = f"{self.base_url}/2.0/repositories/{workspace}/{repo_slug}"
        response = requests.get(repo_url, headers=self.auth_headers)
        response.raise_for_status()
        repo_data = response.json()

        logger.info(f"✅ Repository access confirmed: {repo_data.get('full_name', f'{workspace}/{repo_slug}')}")
        logger.info(f"   Description: {repo_data.get('description') or 'No description'}")
        logger.info(f"   Private: {repo_data.get('is_private', 'Unknown')}")
        logger.info(f"✅ Bitbucket {auth_method} authentication successful")

    except requests.exceptions.RequestException as e:
        logger.info(f"❌ Bitbucket connection failed: {e}")
        logger.info("\nTroubleshooting tips:")
        if auth_method == "Token":
            logger.info("- Verify LAMPE_BITBUCKET_TOKEN is valid and has appropriate permissions")
            logger.info("- Ensure the token has 'repositories:read' scope")
        else:
            logger.info("- Verify LAMPE_BITBUCKET_APP_KEY and LAMPE_BITBUCKET_APP_SECRET are correct")
            logger.info("- Ensure the Bitbucket App is installed on the workspace")
        raise
    except Exception as e:
        logger.info(f"❌ Unexpected error during Bitbucket healthcheck: {e}")
        raise

console

ConsoleProvider(repository: Repository, pull_request: PullRequest)

Bases: Provider

Console provider for delivering PR descriptions to stdout.

Source code in packages/lampe-cli/src/lampe/cli/providers/console.py 
16
17
def __init__(self, repository: Repository, pull_request: PullRequest) -> None:
    super().__init__(repository, pull_request)
deliver_pr_description(payload: PRDescriptionPayload) -> None

Print the PR description to console.

Source code in packages/lampe-cli/src/lampe/cli/providers/console.py 
19
20
21
def deliver_pr_description(self, payload: PRDescriptionPayload) -> None:
    """Print the PR description to console."""
    print(payload.description)
deliver_pr_review(payload: PRReviewPayload) -> None

Print the PR review to console.

Source code in packages/lampe-cli/src/lampe/cli/providers/console.py 
23
24
25
def deliver_pr_review(self, payload: PRReviewPayload) -> None:
    """Print the PR review to console."""
    print(payload.review_markdown)
has_reviewed() -> bool

Check if the token user has already reviewed this PR.

Source code in packages/lampe-cli/src/lampe/cli/providers/console.py 
31
32
33
34
def has_reviewed(self) -> bool:
    """Check if the token user has already reviewed this PR."""
    # Console provider cannot check for existing reviews
    return False
healthcheck() -> None

Check if the console provider is healthy and can connect to the service.

Source code in packages/lampe-cli/src/lampe/cli/providers/console.py 
27
28
29
def healthcheck(self) -> None:
    """Check if the console provider is healthy and can connect to the service."""
    logger.info("✅ Console provider is healthy")

github

GitHubProvider(repository: Repository, pull_request: PullRequest)

Bases: Provider

GitHub provider for delivering PR descriptions to GitHub API.

Source code in packages/lampe-cli/src/lampe/cli/providers/github.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(self, repository: Repository, pull_request: PullRequest) -> None:
    if pull_request.number == 0:
        pr_number = os.getenv("PR_NUMBER")
        if not pr_number:
            raise ValueError("PR_NUMBER environment variable is required for GitHub provider")
        pull_request.number = int(pr_number)

    super().__init__(repository, pull_request)

    # github action has many default environment variables, including the repository full name:
    # https://docs.github.com/en/actions/reference/workflows-and-actions/variables#default-environment-variables
    if repo_name := os.getenv("GITHUB_REPOSITORY"):
        self.owner, self.repo_name = repo_name.split("/")
    else:
        raise ValueError("GITHUB_REPOSITORY environment variable is required for GitHub provider")

    # Initialize GitHub client with appropriate authentication
    self.github_client = self._initialize_github_client()
deliver_pr_description(payload: PRDescriptionPayload) -> None

Update the PR description on GitHub.

Source code in packages/lampe-cli/src/lampe/cli/providers/github.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def deliver_pr_description(self, payload: PRDescriptionPayload) -> None:
    """Update the PR description on GitHub."""
    if self.pull_request.number == 0:
        raise ValueError("Cannot update GitHub PR description for local run")

    try:
        repo = self.github_client.get_repo(f"{self.owner}/{self.repo_name}")
        pull_request = repo.get_pull(self.pull_request.number)
        new_description = update_or_add_text_between_tags(
            pull_request.body or "", payload.description_with_title, "description"
        )
        pull_request.edit(body=new_description)
        logger.info(f"✅ Successfully updated PR #{self.pull_request.number} description on GitHub")
    except Exception as e:
        logger.info(f"❌ Failed to update GitHub PR: {e}")
        # Fallback to console output
        logger.info("Description:")
        logger.info(payload.description)
deliver_pr_review(payload: PRReviewPayload) -> None

Post PR review comments on GitHub.

Source code in packages/lampe-cli/src/lampe/cli/providers/github.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def deliver_pr_review(self, payload: PRReviewPayload) -> None:
    """Post PR review comments on GitHub."""
    if self.pull_request.number == 0:
        raise ValueError("Cannot post GitHub PR review for local run")

    try:
        repo = self.github_client.get_repo(f"{self.owner}/{self.repo_name}")
        pull_request = repo.get_pull(self.pull_request.number)

        # Post review comments for each agent review
        for agent_review in payload.reviews:
            # Post agent summary comment
            if agent_review.summary:
                try:
                    pull_request.create_issue_comment(
                        f"## {agent_review.agent_name}\n\n"
                        f"**Focus Areas:** {', '.join(agent_review.focus_areas)}\n\n"
                        f"{agent_review.summary}"
                    )
                except Exception as e:
                    logger.warning(f"Failed to post agent summary for {agent_review.agent_name}: {e}")

            # Post file-specific comments
            for file_review in agent_review.reviews:
                if file_review.line_comments:
                    # Create review comments for specific lines
                    for line, comment in file_review.line_comments.items():
                        try:
                            # Post a review comment
                            pull_request.create_review_comment(
                                body=f"## 🔦🐛\n{comment}",
                                commit=pull_request.head.sha,
                                path=file_review.file_path,
                                line=int(line),
                            )
                        except Exception as e:
                            logger.warning(f"Failed to post comment for {file_review.file_path}:{line}: {e}")
                            # Fallback: post as general comment
                            pull_request.create_issue_comment(
                                f"**{file_review.file_path} (Line {line}):** {comment}"
                            )

                # Post summary comment if no line comments
                if not file_review.line_comments and file_review.summary:
                    pull_request.create_issue_comment(f"**{file_review.file_path}:** {file_review.summary}")

        logger.info(f"✅ Successfully posted PR #{self.pull_request.number} review comments on GitHub")
    except Exception as e:
        logger.info(f"❌ Failed to post GitHub PR review: {e}")
        # Fallback to console output
        logger.info("Review:")
        logger.info(payload.review_markdown)
has_reviewed() -> bool

Check if the token user has already reviewed this PR.

Source code in packages/lampe-cli/src/lampe/cli/providers/github.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def has_reviewed(self) -> bool:
    """Check if the token user has already reviewed this PR."""
    if self.pull_request.number == 0:
        return False

    try:
        repo = self.github_client.get_repo(f"{self.owner}/{self.repo_name}")
        pull_request = repo.get_pull(self.pull_request.number)

        # Get the authenticated user
        authenticated_user = self.github_client.get_user()

        # Check issue comments (where reviews are posted)
        comments = pull_request.get_issue_comments()
        for comment in comments:
            if comment.user.login == authenticated_user.login:
                return True

        # Also check review comments (inline comments)
        review_comments = pull_request.get_review_comments()
        for comment in review_comments:
            if comment.user.login == authenticated_user.login:
                return True

        return False
    except Exception as e:
        logger.warning(f"Failed to check if PR has been reviewed: {e}")
        return False
healthcheck() -> None

Check if the GitHub provider is healthy and can connect to GitHub.

Source code in packages/lampe-cli/src/lampe/cli/providers/github.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def healthcheck(self) -> None:
    """Check if the GitHub provider is healthy and can connect to GitHub."""
    logger.info("🔍 Checking GitHub provider health...")

    # Check GitHub repository environment variable
    github_repo = os.getenv("GITHUB_REPOSITORY")
    if not github_repo or len(github_repo.split("/")) != 2:
        logger.info("❌ GITHUB_REPOSITORY environment variable not set")
        logger.info("   Set it to 'owner/repo' format (e.g., 'montagne-dev/lampe')")
        raise ValueError("GITHUB_REPOSITORY environment variable not set")
    logger.info(f"✅ GITHUB_REPOSITORY set to: {github_repo}")

    # Check authentication environment variables
    app_id = os.getenv("LAMPE_GITHUB_APP_ID")
    private_key = os.getenv("LAMPE_GITHUB_APP_PRIVATE_KEY")
    token = os.getenv("LAMPE_GITHUB_TOKEN")

    auth_method = None
    if app_id and private_key:
        auth_method = "GitHub App"
        logger.info(f"✅ GitHub App authentication detected (App ID: {app_id})")
    elif token:
        auth_method = "User Token"
        logger.info("✅ User token authentication detected")
    else:
        logger.info("❌ No GitHub authentication found")
        logger.info("   Set either:")
        logger.info("   - LAMPE_GITHUB_APP_ID and LAMPE_GITHUB_APP_PRIVATE_KEY for GitHub App")
        logger.info("   - LAMPE_GITHUB_TOKEN for user token authentication")
        raise ValueError("No GitHub authentication found")

    # Test GitHub connection
    try:
        # Test API access by getting repository info
        repo_info = self.github_client.get_repo(github_repo)
        logger.info(f"✅ Repository access confirmed: {repo_info.full_name}")
        logger.info(f"   Description: {repo_info.description or 'No description'}")
        logger.info(f"   Private: {repo_info.private}")
        logger.info(f"✅ GitHub {auth_method} authentication successful")

    except Exception as e:
        logger.info(f"❌ GitHub connection failed: {e}")
        logger.info("\nTroubleshooting tips:")
        if auth_method == "GitHub App":
            logger.info("- Verify LAMPE_GITHUB_APP_ID and LAMPE_GITHUB_APP_PRIVATE_KEY are correct")
            logger.info("- Ensure the GitHub App is installed on the repository")
            logger.info("- Check that the private key is properly formatted")
        else:
            logger.info("- Verify LAMPE_GITHUB_TOKEN is valid and has appropriate permissions")
            logger.info("- Ensure the token has 'repo' scope for private repositories")
        raise